From 1de4b68f69a5c6a38c25a9abb06b50837f22ed64 Mon Sep 17 00:00:00 2001 From: Daniel Imrie-Situnayake Date: Thu, 12 Mar 2015 12:04:46 -0700 Subject: [PATCH 1/2] Added definitions for kafka-node library --- kafka-node/kafka-node-test.ts | 157 ++++++++++++++++++++++++++++++++++ kafka-node/kafka-node.d.ts | 123 ++++++++++++++++++++++++++ 2 files changed, 280 insertions(+) create mode 100644 kafka-node/kafka-node-test.ts create mode 100644 kafka-node/kafka-node.d.ts diff --git a/kafka-node/kafka-node-test.ts b/kafka-node/kafka-node-test.ts new file mode 100644 index 000000000..607ad5c19 --- /dev/null +++ b/kafka-node/kafka-node-test.ts @@ -0,0 +1,157 @@ +/// + +import kafka = require('kafka-node'); + +var basicClient = new kafka.Client('localhost:2181/', 'sendMessage'); + +var optionsClient = new kafka.Client('localhost:2181/', 'sendMessage', { + sessionTimeout: 30000, + spinDelay: 1000, + retries: 0 +}); +optionsClient.close(); +optionsClient.close(function(){}); + +var producer = new kafka.Producer(basicClient); +producer.on('error', function(error: Error){}); +producer.on('ready', function(){ + + var messages = [{ + topic: 'topicName', + messages: ['message body'], + partition: 0, + attributes: 2 + }, { + topic: 'topicName', + messages: ['message body'], + partition: 0 + }, { + topic: 'topicName', + messages: ['message body'], + attributes: 0 + }, { + topic: 'topicName', + messages: ['message body'] + }, { + topic: 'topicName', + messages: [new kafka.KeyedMessage('key', 'message')] + }]; + + producer.send(messages, function(err: Error){}); + producer.send(messages, function(err: Error, data: Object){}); + + producer.createTopics(['t'], true, function (err: Error, data: Object) {}); + producer.createTopics(['t'], false, function (err, data) {}); + // producer.createTopics(['t'], function (err: Error, data: Object) {}); // Omitting middle argument is not possible in TS + +}); + +var highLevelProducer = new kafka.HighLevelProducer(basicClient); +highLevelProducer.on('error', function(error: Error){}); +highLevelProducer.on('ready', function(){ + + var messages = [{ + topic: 'topicName', + messages: ['message body'], + attributes: 2 + }, { + topic: 'topicName', + messages: ['message body'], + partition: 0 + }, { + topic: 'topicName', + messages: ['message body'], + attributes: 0 + }, { + topic: 'topicName', + messages: ['message body'] + }, { + topic: 'topicName', + messages: [new kafka.KeyedMessage('key', 'message')] + }]; + + producer.send(messages, function(err: Error){}); + producer.send(messages, function(err: Error, data: Object){}); + + producer.createTopics(['t'], true, function (err: Error, data: Object) {}); + producer.createTopics(['t'], false, function (err, data) {}); + // producer.createTopics(['t'], function (err: Error, data: Object) {}); // Omitting middle argument is not possible in TS + +}); + +var fetchRequests = [{ topic: 'awesome' }]; +var consumer = new kafka.Consumer(basicClient, fetchRequests, { + groupId: 'abcde', + autoCommit: true +}); +consumer.on('error', function(error: Error){}); +consumer.on('message', function(message){}); + +consumer.addTopics(['t1', 't2'], function (err, added) {}); +consumer.addTopics([{ topic: 't1', offset: 10 }], function (err, added) {}, true); + +consumer.removeTopics(['t1', 't2'], function (err, removed) {}); + +consumer.commit(function (err, data) {}); + +consumer.setOffset('topic', 0, 0); + +consumer.pause(); +consumer.resume(); +consumer.pauseTopics([ + 'topic1', + { topic: 'topic2', partition: 0 } +]); +consumer.resumeTopics([ + 'topic1', + { topic: 'topic2', partition: 0 } +]); + +consumer.close(true, function () {}); + +var fetchRequests = [{ topic: 'awesome' }]; +var hlConsumer = new kafka.HighLevelConsumer(basicClient, fetchRequests, { + groupId: 'abcde', + autoCommit: true +}); + +hlConsumer.on('error', function(error: Error){}); +hlConsumer.on('message', function(message){}); +hlConsumer.addTopics(['t1', 't2'], function (err, added) {}); +hlConsumer.addTopics([{ topic: 't1', offset: 10 }], function (err, added) {}, true); + +hlConsumer.removeTopics(['t1', 't2'], function (err, removed) {}); + +hlConsumer.commit(function (err, data) {}); + +hlConsumer.setOffset('topic', 0, 0); + +hlConsumer.pause(); +hlConsumer.resume(); +hlConsumer.pauseTopics([ + 'topic1', + { topic: 'topic2', partition: 0 } +]); +hlConsumer.resumeTopics([ + 'topic1', + { topic: 'topic2', partition: 0 } +]); + +hlConsumer.close(true, function () {}); + +var offset = new kafka.Offset(basicClient); + +offset.on('ready', function(){}); + +offset.fetch([ + { topic: 't', partition: 0, time: Date.now(), maxNum: 1 }, + { topic: 't' } +], function (err, data) { }); + +offset.commit('groupId', [ + { topic: 't', partition: 0, offset: 10 } +], function (err, data) { }); + +offset.fetchCommits('groupId', [ + { topic: 't', partition: 0 } +], function (err, data) {}); diff --git a/kafka-node/kafka-node.d.ts b/kafka-node/kafka-node.d.ts new file mode 100644 index 000000000..b90a42a75 --- /dev/null +++ b/kafka-node/kafka-node.d.ts @@ -0,0 +1,123 @@ +// Type definitions for kafka-node 0.2.22 +// Project: https://github.com/SOHU-Co/kafka-node/ +// Definitions by: Daniel Imrie-Situnayake +// Definitions: https://github.com/borisyankov/DefinitelyTyped + +declare module 'kafka-node' { + + // # Classes + export class Client { + constructor(connectionString: string, clientId: string, options?: ZKOptions); + close(callback?: Function): void; + } + + export class Producer { + constructor(client: Client); + on(eventName: string, cb: () => any): void; + on(eventName: string, cb: (error: any) => any): void; + send(payloads: Array, cb: (error: any, data: any) => any): void; + createTopics(topics: Array, async: boolean, cb?: (error: any, data: any) => any): void; + } + + export class HighLevelProducer { + constructor(client: Client); + on(eventName: string, cb: () => any): void; + on(eventName: string, cb: (error: any) => any): void; + send(payloads: Array, cb: (error: any, data: any) => any): void; + createTopics(topics: Array, async: boolean, cb?: (error: any, data: any) => any): void; + } + + export class Consumer { + constructor(client: Client, fetchRequests: Array, options: ConsumerOptions); + on(eventName: string, cb: (message: string) => any): void; + on(eventName: string, cb: (error: any) => any): void; + addTopics(topics: Array, cb: (error: any, added: boolean) => any): void; + addTopics(topics: Array, cb: (error: any, added: boolean) => any, fromOffset: boolean): void; + removeTopics(topics: Array, cb: (error: any, removed: boolean) => any): void; + commit(cb: (error: any, data: any) => any): void; + setOffset(topic: string, partition: number, offset: number): void; + pause(): void; + resume(): void; + pauseTopics(topics: Array /* Array */): void; + resumeTopics(topics: Array /* Array */): void; + close(force: boolean, cb: () => any): void; + } + + export class HighLevelConsumer { + constructor(client: Client, payloads: Array, options: ConsumerOptions); + on(eventName: string, cb: (message: string) => any): void; + on(eventName: string, cb: (error: any) => any): void; + addTopics(topics: Array, cb: (error: any, added: boolean) => any): void; + addTopics(topics: Array, cb: (error: any, added: boolean) => any, fromOffset: boolean): void; + removeTopics(topics: Array, cb: (error: any, removed: boolean) => any): void; + commit(cb: (error: any, data: any) => any): void; + setOffset(topic: string, partition: number, offset: number): void; + pause(): void; + resume(): void; + pauseTopics(topics: Array /* Array */): void; + resumeTopics(topics: Array /* Array */): void; + close(force: boolean, cb: () => any): void; + } + + export class Offset { + constructor(client: Client); + on(eventName: string, cb: () => any): void; + fetch(payloads: Array, cb: (error: any, data: any) => any): void; + commit(groupId: string, payloads: Array, cb: (error: any, data: any) => any): void; + fetchCommits(groupId: string, payloads: Array, cb: (error: any, data: any) => any): void; + } + + export class KeyedMessage { + constructor(key: string, message: string); + } + + // # Interfaces + export interface ZKOptions { + sessionTimeout?: number; + spinDelay?: number; + retries?: number; + } + + export interface ProduceRequest { + topic: string; + messages: any; // Array | Array | string | KeyedMessage + partition?: number; + attributes?: number; + } + + export interface ConsumerOptions { + groupId: string; + autoCommit: boolean; + autoCommitIntervalMs?: number; + fetchMaxWaitMs?: number; + fetchMinBytes?: number; + fetchMaxBytes?: number; + fromOffset?: boolean; + encoding?: string; + } + + export interface Topic { + topic: string; + offset?: number; + } + + export interface OffsetRequest { + topic: string; + partition?: number; + time?: number; + maxNum?: number; + } + + export interface OffsetCommitRequest { + topic: string; + partition?: number; + offset: number; + metadata?: string; + } + + export interface OffsetFetchRequest { + topic: string; + partition?: number; + } + +} From c3a4cd26a4060fdb4f05c57ed1c5fedc42d582c3 Mon Sep 17 00:00:00 2001 From: Daniel Imrie-Situnayake Date: Thu, 12 Mar 2015 13:03:45 -0700 Subject: [PATCH 2/2] Optional properties --- kafka-node/kafka-node.d.ts | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/kafka-node/kafka-node.d.ts b/kafka-node/kafka-node.d.ts index b90a42a75..03105805a 100644 --- a/kafka-node/kafka-node.d.ts +++ b/kafka-node/kafka-node.d.ts @@ -86,8 +86,8 @@ declare module 'kafka-node' { } export interface ConsumerOptions { - groupId: string; - autoCommit: boolean; + groupId?: string; + autoCommit?: boolean; autoCommitIntervalMs?: number; fetchMaxWaitMs?: number; fetchMinBytes?: number;