API Switches

Author: opiepj

README.md

@@ -30,21 +30,21 @@ function connect() {
 
 function events() {
   // Consumer Events
-  consumer.message()
+  consumer.onMessage()
     .subscribe((data) => {
       console.log('Data', data);
     });
-  consumer.error()
+  consumer.onError()
     .subscribe((err) => {
       console.log('Consumer Error', err);
     });
 
   // Producer Events
-  producer.report()
+  producer.onReport()
     .subscribe((report) => {
       console.log('Producer Delivery Report', report);
     });
-  producer.error()
+  producer.onError()
     .subscribe((err) => {
       console.log('Producer Error', err);
     });

sample-override.js

@@ -75,14 +75,14 @@ class SampleOverride {
       this._startConsuming(1);
 
       // Listen to Messages
-      this.consumer.message().subscribe(async (message) => {
+      this.consumer.onMessage().subscribe(async (message) => {
         this._stopConsuming(); // Got a message stop consuming (this will not cancel other consumed messages)
         await this._process(message); // Process Message
         this._startConsuming(1); // Consume the next message
       });
 
       // Logging
-      this.consumer.error().subscribe((err) => {
+      this.consumer.onError().subscribe((err) => {
         console.error(err);
       });
 

sample.js

@@ -9,18 +9,18 @@ const producer = new Producer();
  */
 function consumerEvent(con) {
   return new Promise((resolve, reject) => {
-    con.message()
+    con.onMessage()
       .subscribe((data) => {
         resolve(data);
         console.log('SAMPLE Consumer Data:', data);
       });
-    con.error()
+    con.onError()
       .subscribe((err) => {
         reject(err);
         console.log('SAMPLE Consumer Error:', err);
       });
     console.log('SAMPLE Consumer Connected');
-    con.log()
+    con.onLog()
       .subscribe((log) => {
         console.log('SAMPLE log', log);
       });

src/client.js

@@ -20,18 +20,26 @@ class KafkaClient {
    * Listen to log stream
    * @return {Observable<log>} - log message
    */
-  log() {
+  onLog() {
     return this._logDispatcher.asObservable();
   }
 
   /**
    * Listen to error stream
    * @return {Observable<Error>}
    */
-  error() {
+  onError() {
     return this._errorDispatcher.asObservable();
   }
 
+  /**
+   * Listen to disconnect event
+   * @return {Observable<void>}
+   */
+  onDisconnected() {
+    return this._disconnectDispatcher.asObservable();
+  }
+
   /**
    * Connect to Kafka
    * @protected

src/consumer.js

@@ -56,7 +56,7 @@ class Consumer extends KafkaClient {
    * Message stream to listen to
    * @return {Observable<T>} - message stream
    */
-  message() {
+  onMessage() {
     return this._messageDispatcher.asObservable();
   }
 

src/index.d.ts

@@ -184,9 +184,11 @@ export abstract class Client {
 
   constructor();
 
-  public log(): Observable<Log>;
+  public onLog(): Observable<Log>;
 
-  public error(): Observable<Error>;
+  public onError(): Observable<Error>;
+
+  public onDisconnected(): Observable<void>;
 
   protected connectClient(kafkaClient?: KafkaClient): Promise<{ name: string }>;
 
@@ -205,7 +207,7 @@ export class Consumer extends Client {
 
   constructor(conf?: ConsumerConfig, topicConfig?: TopicConfig);
 
-  public message(): Observable<MessagePayload>;
+  public onMessage(): Observable<MessagePayload>;
 
   public connect(): Promise<{ name: string }>;
 
@@ -227,7 +229,7 @@ export class Producer extends Client {
 
   public publish(message: string | Buffer | number | object | boolean, topic?: string, partition?: number, key?: string, opaque?: string): Promise<DeliveryReport>;
 
-  public report(): Observable<DeliveryReport>;
+  public onReport(): Observable<DeliveryReport>;
 
   public connect(): Promise<{ name: string }>;
 

src/producer.js

@@ -148,7 +148,7 @@ class Producer extends KafkaClient {
    * Stream delivery report from the kafka producer
    * @return {Observable<DeliveryReport>}
    */
-  report() {
+  onReport() {
     return this._deliveryReportDispatcher.asObservable();
   }