// declare type Buffer = any;

declare module "kafka-streams" {

    export interface KafkaHealthConfig {
        thresholds?: {
            consumer?: {
                errors?: number;
                lag?: number;
                stallLag?: number;
                minMessages?: number;
            };
            producer?: {
                errors?: number;
                minMessages?: number;
            };
        };
    }

    export interface CommonKafkaOptions {
        "builtin.features"?: string;
        "client.id"?: string;
        "metadata.broker.list": string;
        "message.max.bytes"?: number;
        "message.copy.max.bytes"?: number;
        "receive.message.max.bytes"?: number;
        "max.in.flight.requests.per.connection"?: number;
        "metadata.request.timeout.ms"?: number;
        "topic.metadata.refresh.interval.ms"?: number;
        "metadata.max.age.ms"?: number;
        "topic.metadata.refresh.fast.interval.ms"?: number;
        "topic.metadata.refresh.fast.cnt"?: number;
        "topic.metadata.refresh.sparse"?: boolean;
        "topic.blacklist"?: string;
        "debug"?: string;
        "socket.timeout.ms"?: number;
        "socket.blocking.max.ms"?: number;
        "socket.send.buffer.bytes"?: number;
        "socket.receive.buffer.bytes"?: number;
        "socket.keepalive.enable"?: boolean;
        "socket.nagle.disable"?: boolean;
        "socket.max.fails"?: number;
        "broker.address.ttl"?: number;
        "broker.address.family"?: "any" | "v4" | "v6";
        "reconnect.backoff.jitter.ms"?: number;
        "statistics.interval.ms"?: number;
        "enabled_events"?: number;
        "log_level"?: number;
        "log.queue"?: boolean;
        "log.thread.name"?: boolean;
        "log.connection.close"?: boolean;
        "internal.termination.signal"?: number;
        "api.version.request"?: boolean;
        "api.version.fallback.ms"?: number;
        "broker.version.fallback"?: string;
        "security.protocol"?: "plaintext" | "ssl" | "sasl_plaintext" | "sasl_ssl";
        "ssl.cipher.suites"?: string;
        "ssl.key.location"?: string;
        "ssl.key.password"?: string;
        "ssl.certificate.location"?: string;
        "ssl.ca.location"?: string;
        "ssl.crl.location"?: string;
        "sasl.mechanisms"?: string;
        "sasl.kerberos.service.name"?: string;
        "sasl.kerberos.principal"?: string;
        "sasl.kerberos.kinit.cmd"?: string;
        "sasl.kerberos.keytab"?: string;
        "sasl.kerberos.min.time.before.relogin"?: number;
        "sasl.username"?: string;
        "sasl.password"?: string;
        "partition.assignment.strategy"?: string;
        "session.timeout.ms"?: number;
        "heartbeat.interval.ms"?: number;
        "group.protocol.type"?: string;
        "coordinator.query.interval.ms"?: number;
        "event_cb"?: boolean;
        "group.id": string;
        "enable.auto.commit"?: boolean;
        "auto.commit.interval.ms"?: number;
        "enable.auto.offset.store"?: boolean;
        "queued.min.messages"?: number;
        "queued.max.messages.kbytes"?: number;
        "fetch.wait.max.ms"?: number;
        "fetch.message.max.bytes"?: number;
        "fetch.min.bytes"?: number;
        "fetch.error.backoff.ms"?: number;
        "offset.store.method"?: "none" | "file" | "broker";
        "enable.partition.eof"?: boolean;
        "check.crcs"?: boolean;
        "queue.buffering.max.messages"?: number;
        "queue.buffering.max.kbytes"?: number;
        "queue.buffering.max.ms"?: number;
        "message.send.max.retries"?: number;
        "retry.backoff.ms"?: number;
        "compression.codec"?: "none" | "gzip" | "snappy" | "lz4";
        "batch.num.messages"?: number;
        "delivery.report.only.error"?: boolean;
    }

    export interface CombinedKafkaConfig {
        kafkaHost?: string;
        clientName?: string;
        groupId?: string;
        workerPerPartition?: number;
        options?: {
            sessionTimeout?: number;
            protocol?: [string];
            fromOffset?: string;
            fetchMaxBytes?: number;
            fetchMinBytes?: number;
            fetchMaxWaitMs?: number;
            heartbeatInterval?: number;
            retryMinTimeout?: number;
            requireAcks?: number;
            ackTimeoutMs?: number;
            partitionerType?: number;
        };
        health?: KafkaHealthConfig;
        tconf?: {
            "auto.commit.enable"?: boolean;
            "auto.commit.interval.ms"?: number;
            "auto.offset.reset"?: "smallest" | "earliest" | "beginning" | "largest" | "latest" | "end" | "error";
            "offset.store.path"?: string;
            "offset.store.sync.interval.ms"?: number;
            "offset.store.method"?: "file" | "broker";
            "consume.callback.max.messages"?: number;
            "request.required.acks"?: number;
            "request.timeout.ms"?: number;
            "message.timeout.ms"?: number;
            "produce.offset.report"?: boolean;
        };
        noptions?: CommonKafkaOptions;
    }

    export interface KafkaStreamsConfig extends CombinedKafkaConfig {
        batchOptions?: {
            batchSize: number;
            commitEveryNBatch: number;
            concurrency: number;
            commitSync: boolean;
            noBatchCommits: boolean;
        };
    }

    export class KafkaClient { }
    export class JSKafkaClient extends KafkaClient { }
    export class NativeKafkaClient extends KafkaClient { }

    export interface Storage {
        set(key: any, value: any): Promise<any>;
        get(key: any): Promise<any>;
        increment(key: any, by?: number): Promise<number>;
        getState(): Promise<object>;
        setState(newState: object): Promise<boolean>;
        close(): Promise<boolean>;
    }

    export class KStorage implements Storage {
        constructor(options: object);
        set(key: any, value: any): Promise<any>;
        get(key: any): Promise<any>;
        increment(key: any, by?: number): Promise<number>;
        getState(): Promise<object>;
        setState(newState: object): Promise<boolean>;
        close(): Promise<boolean>;
    }

    export class Window {
        constructor(container: any[], collect: boolean);
        getStream(): KStream;
        execute(element: any, leaveEncapsulated: boolean): void;
        writeToStream(): void;
    }

    export class StreamDSL {
        constructor(topicName: string, storage?: KStorage, kafka?: KafkaClient, isClone?: boolean);
        getStats(): any;
        getStorage(): KStorage;
        writeToStream(message: any | Array<any>): void;
        getMost(): any;
        getNewMostFrom(array: Array<any>): any;
        replaceInternalObservable(newStream$: any): void;
        setProduceHandler(handler: any): void;
        createAndSetProduceHandler(): any;
        setKafkaStreamsReference(reference: KafkaStreams): void;
        from(topic: string | Array<string>): StreamDSL;
        awaitPromises(etl: (value: Promise<any>) => any): StreamDSL;
        map(etl: (value: any) => any): StreamDSL;
        asyncMap(etl: (value: any) => Promise<any>): StreamDSL;
        concatMap(etl: (valueForStream: any) => any): StreamDSL;
        forEach(etl: (value: any) => void): Promise<void>;
        chainForEach(etl: Function, callback: Function): StreamDSL;
        tap(etl: (value: any) => void): StreamDSL;
        filter(pred: (value: any) => boolean): StreamDSL;
        skipRepeats(): StreamDSL;
        skipRepeatsWith(equals: Function): StreamDSL;
        skip(count: number): StreamDSL;
        take(count: number): StreamDSL;
        multicast(): StreamDSL;
        mapStringToArray(delimiter: string): StreamDSL;
        mapArrayToKV(keyIndex: number, valueIndex: number): StreamDSL;
        mapStringToKV(delimiter: string, keyIndex: number, valueIndex: number): StreamDSL;
        mapJSONParse(): StreamDSL;
        mapStringify(): StreamDSL;
        mapBufferKeyToString(): StreamDSL;
        mapBufferValueToString(): StreamDSL;
        mapStringValueToJSONObject(): StreamDSL;
        mapJSONConvenience(): StreamDSL;
        wrapAsKafkaValue(topic?: string): StreamDSL;
        mapWrapKafkaValue(): StreamDSL;
        atThroughput(count: number, callback: (message: any) => void): StreamDSL;
        mapToFormat(type: string, getId?: (message: any) => any): StreamDSL;
        mapFromFormat(): StreamDSL;
        timestamp(etl: (message: any) => any): StreamDSL;
        constant(substitute: any): StreamDSL;
        scan(eff: (val: any, ret: any) => any, initial: any): StreamDSL;
        slice(start: number, end: number): StreamDSL;
        takeWhile(pred: (message: any) => boolean): StreamDSL;
        skipWhile(pred: (message: any) => boolean): StreamDSL;
        until(signal$: any): StreamDSL;
        since(signal$: any): StreamDSL;
        continueWith(f: () => any): StreamDSL;
        reduce(eff: (val: any, ret: any) => any, initial: any): Promise<any>;
        chainReduce(eff: (val: any, ret: any) => any, initial: any, callback: (error: Error, value: any) => void): StreamDSL;
        drain(): Promise<void>;
        throttle(throttlePeriod: number): StreamDSL;
        delay(delayTime: number): StreamDSL;
        debounce(debounceTime: number): StreamDSL;
        countByKey(key: string, countFieldName?: string): StreamDSL;
        sumByKey(key: string, fieldName: string, sumField: boolean): StreamDSL;
        min(fieldName: string, minField?: string): StreamDSL;
        max(fieldName: string, maxField?: string): StreamDSL;
        to(topic?: string, outputPartitionsCount?: number | "auto", produceType?: "send" | "buffer" | "bufferFormat",
            version?: number, compressionType?: number, producerErrorCallback?: (error: Error) => void,
            outputKafkaConfig?: KafkaStreamsConfig): Promise<boolean>;
    }

    export class KStream extends StreamDSL {
        constructor(topicName: string, storage?: KStorage, kafka?: KafkaClient, isClone?: boolean);
        start(kafkaReadyCallback?: () => void, kafkaErrorCallback?: (error: Error) => void,
            withBackPressure?: boolean, outputKafkaConfig?: KafkaStreamsConfig): Promise<void>;
        start(config: { outputKafkaConfig?: KafkaStreamsConfig }): Promise<void>;
        innerJoin(stream: KStream, key?: string, windowed?: boolean, combine?: Function): KStream;
        merge(stream: KStream): KStream;
        fromMost(stream$: any): KStream;
        clone(): KStream;
        window(from: number, to: number, etl?: null | ((message: any) => any), encapsulated?: boolean): { window: Window, abort: () => void, stream: KStream };
        branch(preds: ((message: any) => boolean)[]): KStream[];
        close(): Promise<boolean>;
    }

    export class KTable extends StreamDSL {
        constructor(topicName: string, keyMapETL: (message: any) => { key: any, value: any }, storage?: KStorage, kafka?: KafkaClient, isClone?: boolean);
        start(kafkaReadyCallback?: () => void, kafkaErrorCallback?: (error: Error) => void,
            withBackPressure?: boolean, outputKafkaConfig?: KafkaStreamsConfig): Promise<void>;
        writeToTableStream(message: any): void;
        finalise(buildReadyCallback: () => void): void;
        consumeUntilMs(ms: number, finishedCallback: () => void): void;
        consumeUntilCount(count: number, finishedCallback: () => void): void;
        getTable(): object;
        replay(): void;
        merge(stream: StreamDSL): Promise<KTable>;
        clone(): KTable;
        close(): Promise<boolean>;
    }

    export class KafkaStreams {

        constructor(config: KafkaStreamsConfig, storageClass?: new () => KStorage, storageOptions?: object,
            disableStorageTest?: boolean);

        static checkStorageClass(storageClass: new () => KStorage): void;
        getKafkaClient(topic: string): KafkaClient;
        getStorage(): KStorage;
        getKStream(topic?: string, storage?: new () => KStorage): KStream;
        fromMost($stream: any, storage?: new () => KStorage): KStream;
        getKTable(topic: string, keyMapETL: Function, storage: new () => KStorage): KTable;
        getStats(): Array<object>;
        closeAll(): Promise<Array<boolean>>;
    }
}