<commit_before>//
//  KafkaProducer.swift
//  SwiftKafka
//
//  Created by <NAME> on 26/08/2017.
//
//

import Foundation

import ckafka

class KafkaProducer: KafkaBase {
 
    // MARK: - Public Properties
    
    /// The number of messages waiting to be delivered
    /// to the broker
    public var numOfPendingMessages: Int32? {
        
        guard let h = handle else {
            return nil
        }
        
        return rd_kafka_outq_len(h)
        
    }
    
    // MARK: - Initialiser
    
    public init(globalConfiguration: GlobalConfig? = nil, topicConfiguration: TopicConfig? = nil) throws {
        
        let globalConfig = try (globalConfiguration ?? (try GlobalConfig()))
        let topicConfig = try (topicConfiguration ?? (try TopicConfig()))
        
        globalConfig.configureMessageDeliveryCallback()
        
        try super.init(withClientType: .producer, globalConfig: globalConfig, andTopicConfig: topicConfig)
        
    }
    
    deinit {
        let _ = try? flush(1000)
    }
    
    // MARK: - Public Methods
    
    /**
     Produce message to topic. This operation is asynchronous and you may want to pass in
     `deliveryClosure` that will be called from `poll()` when the message has been deliverd or permanently failed.
     - parameter topic: The name of the topic
     - parameter key: The message key
     - parameter value: The message value
     - parameter partition: The partition to produce too (defaults to default partitioner)
     - parameter deliveryClosure: A closure to execute if the message gets delivered or permanently failed.
    */<commit_msg>Add optional producer-level message delivery callback<commit_after>//
//  KafkaProducer.swift
//  SwiftKafka
//
//  Created by <NAME> on 26/08/2017.
//
//

import Foundation

import ckafka

class KafkaProducer: KafkaBase {
 
    // MARK: - Public Properties
    
    /// The number of messages waiting to be delivered
    /// to the broker
    public var numOfPendingMessages: Int32? {
        
        guard let h = handle else {
            return nil
        }
        
        return rd_kafka_outq_len(h)
        
    }
    
    /// A closure to execute when a message is delivered
    public var onMessageDelivery: MessageDeliveryClosure?
    
    // MARK: - Initialiser
    
    public init(globalConfiguration: GlobalConfig? = nil, topicConfiguration: TopicConfig? = nil) throws {
        
        let globalConfig = try (globalConfiguration ?? (try GlobalConfig()))
        let topicConfig = try (topicConfiguration ?? (try TopicConfig()))
        
        globalConfig.configureMessageDeliveryCallback()
        
        try super.init(withClientType: .producer, globalConfig: globalConfig, andTopicConfig: topicConfig)
        
    }
    
    deinit {
        let _ = try? flush(1000)
    }
    
    // MARK: - Public Methods
    
    /**
     Produce message to topic. This operation is asynchronous and you may want to pass in
     `deliveryClosure` that will be called from `poll()` when the message has been deliverd or permanently failed.
     - parameter topic: The name of the topic
     - parameter key: The message key
     - parameter value: The message value
     - parameter partition: The partition to produce too (defaults to default partitioner)
     - parameter deliveryClosure: A closure to execute if the message gets delivered or permanently failed.
    */
{'fraction_non_alphanumeric': 0.0750733137829912, 'fraction_numerical': 0.008211143695014663, 'mean_word_length': 2.2892960462873675, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '2388386', 'n_tokens_mistral': 928, 'n_tokens_neox': 882, 'n_words': 322}