#! /usr/bin/env python3
# -*- coding: utf-8 -*-

# `pip install …`
import kafka  # … kafka-python

import argparse
from concurrent.futures import ThreadPoolExecutor
import enum
import multiprocessing
import sys
import time


class Sync(enum.Enum):
    NONE = 'none'
    LEAD = 'leader'
    ALL = 'all'

    def __str__(self):
        return self.value

    def convert(self):
        values = {
            str(Sync.NONE): '0',
            str(Sync.LEAD): '1',
            str(Sync.ALL): 'all',
        }
        return values[self.value]


def main():
    parser = argparse.ArgumentParser(description='Produce a single message taken from input')
    parser.add_argument('--server', type=str, metavar='HOST', default='localhost',
        help='Kafka bootstrap-server address')
    parser.add_argument('--port', type=int, metavar='PORT', default=9092,
        help='Kafka bootstrap-server port')
    parser.add_argument('--client', type=str, default='ch-kafka-python',
        help='custom client id for this producer')
    parser.add_argument('--topic', type=str, required=True,
        help='name of Kafka topic to store in')
    parser.add_argument('--retries', type=int, default=0,
        help='number of retries to send on failure')
    parser.add_argument('--multiply', type=int, default=1,
        help='multiplies incoming string many times')
    parser.add_argument('--repeat', type=int, default=1,
        help='send same (multiplied) message many times')

    mode_group = parser.add_mutually_exclusive_group()
    mode_group.add_argument('--jobs', type=int, default=multiprocessing.cpu_count(),
        help='number of concurrent jobs')
    mode_group.add_argument('--delay', type=int, metavar='SECONDS', default=0,
        help='delay before sending next message')

    args = parser.parse_args()
    config = {
        'bootstrap_servers': f'{args.server}:{args.port}',
        'client_id': args.client,
        'retries': args.retries,
    }
    client = kafka.KafkaProducer(**config)

    message = sys.stdin.buffer.read() * args.multiply

    def send(num):
        if args.delay > 0:
            time.sleep(args.delay)
        client.send(topic=args.topic, value=message)
        print(f'iteration {num}: sent a message multiplied {args.multiply} times')

    if args.delay > 0:
        args.jobs = 1
    pool = ThreadPoolExecutor(max_workers=args.jobs)
    for num in range(args.repeat):
        pool.submit(send, num)
    pool.shutdown()

    client.flush()
    client.close()
    return 0


if __name__ == "__main__":
    exit(main())