kellanburket / franz Goto Github PK

import Foundation
import ReactiveCocoa

@objc class BioKafkaInterface: NSObject, BioCommunicator, HighLevelConsumerDelegate {

let cluster : Cluster


init(ip:String, port:Int, clientId:String, groupId:String, partition:Int, topic:String) {
    print("init ip string")
    self.cluster = Cluster(brokers:[(ip, Int32(port))], clientId:clientId);
    super.init()
    self.cluster.getHighLevelConsumer(topic, partition: Int32(partition), groupId: groupId, delegate:self)

}

func read(size:UInt) -> Signal<[Int8], NSError> {
    return Signal.empty
}

func write(data:[Int8]) -> SignalProducer<Int, NSError> {
    return SignalProducer.empty;
}

func consumerDidReturnMessage(message: Message, offset: Int64) {
    print(NSString(data: message.value, encoding: NSUTF8StringEncoding))
}

func consumerIsReady(consumer: Consumer) {
    print("Consummer is ready")
    if let highLevelConsumer = consumer as? HighLevelConsumer {
        print("\(highLevelConsumer.broker?.description)")
        highLevelConsumer.poll()
    }
}


func shouldCommitOffset(topic: String, partition: Int32, offset: Int64) -> Bool {
    print("shouldCommitOffset \(topic, partition, offset)")
    return true
}

func shouldAttachOffsetMetadata(topic: String, partition: Int32, offset: Int64) -> String? {
    print("shouldAttachOffsetMetadata \(topic, partition, offset)")
    return "toto"
}

func offsetDidCommit(topic: String, partition: Int32, offset: Int64) {
    print("offsetDidCommit \(topic, partition, offset)")
}

func offsetCommitDidFail(topic: String, partition: Int32, offset: Int64, errorId: Int16, errorDescription: String) {
    print("offsetCommitDidFail \(topic, partition, offset, errorId, errorDescription)")
}


func shouldRetryFailedOffsetCommit(topic: String, partition: Int32, offset: Int64, errorId: Int16, errorDescription: String) -> Bool {
    print("shouldRetryFailedOffsetCommit \(topic, partition, offset, errorId, errorDescription)")
    return true;
}



func fetchDidFail(
    topic: String,
    partition: Int32,
    errorId: Int16,
    errorDescription: String
    ) {
        print("fetchDidFail \(topic, partition, errorId, errorDescription)");
}

func shouldRetryFailedFetch(
    topic: String,
    partition: Int32,
    errorId: Int16,
    errorDescription: String
    ) -> Bool {
                    print("shouldRetryFailedFetch \(topic, partition, errorId, errorDescription)");
        return true
}

func topicPartitionLeaderNotFound(topic: String, partition: Int32) {
    print("topicPartitionLeaderNotFound \(topic, partition) ");
}

init() {
    self.cluster = Cluster(brokers: [("localhost", 9092)], clientId: "data")
    self.consumer = self.cluster.getConsumer(topics: ["data-grid"], groupId: "data")

    self.cluster.sendMessage("data-grid", message: "Grid data")
    print("hello")
    self.consumer.listen { [weak self] message in
        DispatchQueue.main.async {
            if let messageString = String(data: message.value, encoding: .utf8) {
                print(messageString)
                self?.receivedMessages.append(messageString)
            } else {
                print("Error: Failed to decode message data as UTF-8 string")
            }
        }
    }


    // No need to subscribe separately; topics are provided during consumer creation
    // Example of sending a message (move this to appropriate place)