         
     }
     
+    // MARK: - Public Methods
+    
+    /**
+     Adds one or more brokers to the kafka handle's list of initial bootstrap brokers.
+     - Additional brokers will be discovered automatically as soon as rdkafka connects to a broker by querying the broker metadata.
+     - If a broker name resolves to multiple addresses (and possibly address families) all will be used for connection attempts in round-robin fashion.
+     - parameter brokers: A list of bootstrap brokers to connect to
+     - returns: The number of brokers that were successfully added
+    */
+    public func connect(toBrokers brokers: [BrokerConnection]) -> Int {
+        
+        let brokersList = brokers.map({ $0.description }).joined(separator: ",")
+        
+        return Int(rd_kafka_brokers_add(kafkaClientHandle, brokersList))
+        
+    }
+    
+    // MARK: - Interal Methods
+    internal func getMetadata(forTopicHandle topicHandle: OpaquePointer? = nil,
+                              timeout: Int32 = 1000) throws -> Metadata
+    {
+        
+        guard let kafkaClientHandle = self.kafkaClientHandle else {
+            throw SwiftKafkaError.unknownError
+        }
+        
+        let ppMetadata = UnsafeMutablePointer<UnsafePointer<rd_kafka_metadata>?>.allocate(capacity: 1)
+        var error: rd_kafka_resp_err_t
+        
+        defer {
+            
+            rd_kafka_metadata_destroy(ppMetadata.pointee)
+            ppMetadata.deallocate(capacity: 1)
+            
+        }
+        
+        if let topicHandle = topicHandle {
+            error = rd_kafka_metadata(kafkaClientHandle, 0, topicHandle, ppMetadata, timeout)
+        } else {
+            error = rd_kafka_metadata(kafkaClientHandle, 1, nil, ppMetadata, timeout)
+        }
+        
+        guard error.rawValue == 0 else {
+            let coreError = SwiftKafkaCoreError(rdError: error)
+            throw SwiftKafkaError.coreError(coreError)
+        }
+        
+        guard let rawMetadata = (ppMetadata.pointee)?.pointee else {
+            throw SwiftKafkaError.unknownError
+        }
+        
+        return Metadata.metadata(fromRawMetadata: rawMetadata)
+        
+    }
+    
 }

        return String(cString: rd_kafka_version_str())
    }
    
    // MARK: - Private Properties
    public private(set) var kafkaClientHandle: OpaquePointer?
    
    // MARK: - Initialiser
    public init(withClientType clientType: KafkaClientType,
                andConfig config: KafkaConfig? = nil) throws
    {
        
    }
    

import ckafka

let kSwiftKafkaCStringSize = 1024

/// Enum representing the possible Kafka
/// client types
