kylefang/DefinitelyTyped
1
0
Fork 0
mirror of https://github.com/zhigang1992/DefinitelyTyped.git synced 2026-05-23 04:47:24 +08:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #14364 from bkim54/update-kafka-node-typings

Browse Source 
Updating kafka-node typings with ConsumerGroup definitions
This commit is contained in:
Arthur Ozga
2017-02-10 13:22:10 -08:00
committed by GitHub
parent 41cf416171 c4ed2ff42e
commit 6d114d856e
2 changed files with 69 additions and 19 deletions
Download Patch File Download Diff File Expand all files Collapse all files

66
kafka-node/index.d.ts vendored
View File

@@ -1,4 +1,4 @@
// Type definitions for kafka-node 0.2.22
// Type definitions for kafka-node 1.3.3
// Project: https://github.com/SOHU-Co/kafka-node/
// Definitions by: Daniel Imrie-Situnayake <https://github.com/dansitu/>
// Definitions: https://github.com/DefinitelyTyped/DefinitelyTyped
@@ -13,24 +13,24 @@ export declare class Client {
export declare class Producer {
constructor(client: Client);
on(eventName: string, cb: () => any): Producer;
on(eventName: string, cb: (error: any) => any): Producer;
on(eventName: string, cb: () => any): void;
on(eventName: string, cb: (error: any) => any): void;
send(payloads: Array<ProduceRequest>, cb: (error: any, data: any) => any): void;
createTopics(topics: Array<string>, async: boolean, cb?: (error: any, data: any) => any): void;
}
export declare class HighLevelProducer {
constructor(client: Client);
on(eventName: string, cb: () => any): HighLevelProducer;
on(eventName: string, cb: (error: any) => any): HighLevelProducer;
constructor(client: Client, options?: any);
on(eventName: string, cb: () => any): void;
on(eventName: string, cb: (error: any) => any): void;
send(payloads: Array<ProduceRequest>, cb: (error: any, data: any) => any): void;
createTopics(topics: Array<string>, async: boolean, cb?: (error: any, data: any) => any): void;
}
export declare class Consumer {
constructor(client: Client, fetchRequests: Array<FetchRequest>, options: ConsumerOptions);
on(eventName: string, cb: (message: string) => any): Consumer;
on(eventName: string, cb: (error: any) => any): Consumer;
constructor(client: Client, fetchRequests: Array<OffsetFetchRequest>, options: ConsumerOptions);
on(eventName: string, cb: (message: string) => any): void;
on(eventName: string, cb: (error: any) => any): void;
addTopics(topics: Array<string>, cb: (error: any, added: boolean) => any): void;
addTopics(topics: Array<Topic>, cb: (error: any, added: boolean) => any, fromOffset: boolean): void;
removeTopics(topics: Array<string>, cb: (error: any, removed: boolean) => any): void;
@@ -45,8 +45,8 @@ export declare class Consumer {
export declare class HighLevelConsumer {
constructor(client: Client, payloads: Array<Topic>, options: ConsumerOptions);
on(eventName: string, cb: (message: string) => any): HighLevelConsumer;
on(eventName: string, cb: (error: any) => any): HighLevelConsumer;
on(eventName: string, cb: (message: string) => any): void;
on(eventName: string, cb: (error: any) => any): void;
addTopics(topics: Array<string>, cb: (error: any, added: boolean) => any): void;
addTopics(topics: Array<Topic>, cb: (error: any, added: boolean) => any, fromOffset: boolean): void;
removeTopics(topics: Array<string>, cb: (error: any, removed: boolean) => any): void;
@@ -59,14 +59,22 @@ export declare class HighLevelConsumer {
close(force: boolean, cb: () => any): void;
}
export declare class ConsumerGroup {
constructor(options: ConsumerGroupOptions, topics: string[]);
on(eventName: string, cb: (message: string) => any): void;
on(eventName: string, cb: (error: any) => any): void;
close(force: boolean, cb: (error: any) => any): void;
}
export declare class Offset {
constructor(client: Client);
on(eventName: string, cb: () => any): Offset;
on(eventName: string, cb: () => any): void;
fetch(payloads: Array<OffsetRequest>, cb: (error: any, data: any) => any): void;
commit(groupId: string, payloads: Array<OffsetCommitRequest>, cb: (error: any, data: any) => any): void;
fetchCommits(groupId: string, payloads: Array<OffsetFetchRequest>, cb: (error: any, data: any) => any): void;
fetchLatestOffsets(topics: Array<string>, cb: (error: any, data: any) => any): void;
on(eventName: string, cb: (error: any) => any): Offset;
fetchEarliestOffsets(topics: Array<string>, cb: (error: any, data: any) => any): void;
on(eventName: string, cb: (error: any) => any): void;
}
export declare class KeyedMessage {
@@ -74,6 +82,11 @@ export declare class KeyedMessage {
}
// # Interfaces
export interface AckBatchOptions {
noAckBatchSize: number | null,
noAckBatchAge: number | null
}
export interface ZKOptions {
sessionTimeout?: number;
spinDelay?: number;
@@ -90,6 +103,7 @@ export interface ProduceRequest {
export interface ConsumerOptions {
groupId?: string;
id?: string;
autoCommit?: boolean;
autoCommitIntervalMs?: number;
fetchMaxWaitMs?: number;
@@ -99,6 +113,27 @@ export interface ConsumerOptions {
encoding?: string;
}
export interface CustomPartitionAssignmentProtocol {
name: string;
version: number;
userData: {};
assign: (topicPattern: any, groupMembers: any, callback: (error: any, result: any) => void) => void;
}
export interface ConsumerGroupOptions {
host: string;
zk?: ZKOptions;
batch?: AckBatchOptions;
ssl?: boolean;
id: string;
groupId: string;
sessionTimeout: number;
protocol: Array<"roundrobin" | "range" | CustomPartitionAssignmentProtocol>;
fromOffset: "earliest" | "latest" | "none";
migrateHLC: false;
migrateRolling: true;
}
export interface Topic {
topic: string;
offset?: number;
@@ -123,10 +158,5 @@ export interface OffsetCommitRequest {
export interface OffsetFetchRequest {
topic: string;
partition?: number;
}
export interface FetchRequest {
topic: string;
offset?: number;
}

22
kafka-node/kafka-node-tests.ts
View File

@@ -1,4 +1,3 @@
import kafka = require('kafka-node');
var basicClient = new kafka.Client('localhost:2181/', 'sendMessage');
@@ -138,6 +137,24 @@ hlConsumer.resumeTopics([
hlConsumer.close(true, function () {});
var ackBatchOptions = {'noAckBatchSize': 1024, 'noAckBatchAge': 10};
var cgOptions: kafka.ConsumerGroupOptions = {
host: 'localhost:2181/',
batch: ackBatchOptions,
groupId: 'groupID',
id: 'consumerID',
sessionTimeout: 15000,
protocol: ["roundrobin"],
fromOffset: "latest",
migrateHLC: false,
migrateRolling: true
};
var consumerGroup = new kafka.ConsumerGroup( cgOptions, ['topic1']);
consumerGroup.on('error', (err) => {});
consumerGroup.on('message', (msg) => {});
consumerGroup.close(true, () => {});
var offset = new kafka.Offset(basicClient);
offset.on('ready', function(){});
@@ -154,3 +171,6 @@ offset.commit('groupId', [
offset.fetchCommits('groupId', [
{ topic: 't', partition: 0 }
], function (err, data) {});
offset.fetchLatestOffsets(['t'], (err, offsets) => {})
offset.fetchEarliestOffsets(['t'], (err, offsets) => {})