Merge pull request #3 from botorjs/create_channel

create channel

README.md

@@ -1,2 +1,63 @@
-# event-bus
-Library that core of Botorjs is a event bus
+# Event Bus
+[![Build Status](https://travis-ci.org/botorjs/event-bus.svg?branch=master)](https://travis-ci.org/botorjs/event-bus)
+[![Coverage Status](https://coveralls.io/repos/github/botorjs/event-bus/badge.svg?branch=master)](https://coveralls.io/github/botorjs/event-bus?branch=master)
+
+### Library that core of Botorjs is a event bus
+
+# Installation
+```
+npm install @botorjs/event-bus --save
+```
+
+# Setup and Example
+
+
+```js
+
+    import { EventBus } from "../src/EventBus";
+    var bus: EventBus = new EventBus();
+
+    // 
+    bus.on("test", function(data) {
+       // process
+    })
+    bus.emit("test", "test");
+
+    // channel point-to-point
+    bus.registerChannel("test", "test", (data: ContextChannel) => {
+        // process
+    });
+    bus.emit("test","t");
+    bus.emit("test","t");
+    bus.emit("test","t");
+```
+
+# API
+
+## EventBus
+
+| Property   |      Description      |
+|---------- |:-------------|
+| on(name, callback)  |  listen event | 
+| once(name, callback) |    listen event one time  | 
+| off(name, callback) | remove listen event | 
+| registerChannel(name, event_name, callback, limit = 0) |  register a channel with callback handle |
+| getChannel(name)  |  get channel have register | 
+| removeChannel(name)  |  remove channel have register | 
+
+
+## ContextChannel
+* data connext of Channel
+
+| Property   |      Description      |
+|---------- |:-------------|
+| data  |  get value data | 
+| channel |   set channel event  | 
+
+## Channel
+
+| Property   |      Description      |
+|---------- |:-------------|
+| name  |  name channel |
+| listen |   listen event  |
+| remove |   remove listen event  |

index.ts

@@ -0,0 +1,5 @@
+import { Channel  } from './src/Channel';
+import { EventBus } from './src/EventBus';
+import { ContextChannel } from './src/ContextChannel';
+
+export { Channel, EventBus, ContextChannel }

package.json

@@ -20,7 +20,8 @@
     "event-bus",
     "event",
     "bus",
-    "chanel",
+    "channel",
+    "point-to-point",
     "node",
     "typescript"
   ],

src/Channel.ts

@@ -0,0 +1,88 @@
+import { EventEmitter } from 'events';
+import { ContextChannel } from './ContextChannel';
+import { EventBus } from './EventBus';
+import  Debug from 'debug';
+
+var debug = Debug("channel");
+
+export enum StateChannel {
+    Process,
+    Waiting,
+}
+
+export class Channel {
+
+    private _name: string;
+    private _state: StateChannel;
+    private _limist_queue: number;
+    private _queue: any[];
+    private _bus: EventEmitter;
+    private _callback: Function;
+    private _context: ContextChannel;
+    private _event_name: string;
+    private _event_bus: EventBus;
+
+    constructor(event_bus: EventBus, name: string,event_name: string, callback: Function, limit: number = 0) {
+        this._event_bus = event_bus;
+        this._event_name = event_name;
+        this._name = name;
+        this._callback = callback;
+        this._limist_queue = limit;
+        this._bus = new EventEmitter();
+        this._queue = new Array();
+        this._bus.on("next", this._next.bind(this));
+        this._bus.on("handler", this.handler.bind(this));
+        this._context = new ContextChannel(this);
+
+        //
+        this._state = StateChannel.Waiting;
+    }
+
+    public get name(): string {
+        return this._name;
+    }
+
+    private handler(data: any) {
+        this._state = StateChannel.Process;
+        this._context.data = data;
+        try {
+            this._callback(this._context);
+        } finally {
+            this._state = StateChannel.Waiting;
+        }
+        this.callEventNext();
+    }
+
+    private callEventNext() {
+        debug("call next event");
+        this._bus.emit("next");
+    }
+
+    public listen() {
+        this._event_bus.on(this._event_name, this._listen.bind(this));
+    }
+
+    public remove() {
+        this._event_bus.off(this._event_name, this._listen.bind(this));
+    }
+
+    private _listen(data: any) {
+        debug("listen new event channel");
+        if(this._limist_queue != 0 && this._limist_queue >= this._queue.length) {
+            return;
+        }
+        this._queue.push(data);
+        this.callEventNext();
+    }
+
+    private _next() {
+        debug("next process");
+        if(this._state == StateChannel.Waiting) {
+            debug("ext process StateChannel Waiting");
+            if(this._queue.length > 0) {
+                var data = this._queue.shift();
+                this._bus.emit("handler", data)
+            }
+        }
+    }
+}

src/ContextChannel.ts

@@ -0,0 +1,15 @@
+import { Channel } from "./Channel";
+
+
+export class ContextChannel {
+    public data: any;
+    private _channel: Channel;
+
+    constructor(channel: Channel) {
+        this._channel = channel;
+    }
+
+    public get channel():Channel{
+        return this._channel;
+    }
+}

src/EventBus.ts

@@ -1,10 +1,13 @@
 import  { EventEmitter } from 'events'
+import { Channel } from './Channel';
 
 export class EventBus {
     private _event_bus: EventEmitter
+    private _channel: Map<string, Channel>;
 
     constructor() {
         this._event_bus = new EventEmitter();
+        this._channel = new Map<string, Channel>();
     }
 
     /**
@@ -13,7 +16,7 @@ export class EventBus {
      * @param name name
      * @param callback callback when listen event name
      */
-    on(name: string, callback: (...args: any[]) => void) {
+    public on(name: string, callback: (...args: any[]) => void) {
         this._event_bus.on(name, callback);
     }
 
@@ -23,11 +26,38 @@ export class EventBus {
      * @param name event name
      * @param data data event
      */
-    emit(name: string, data: any = null) {
+    public emit(name: string, data: any = null) {
         this._event_bus.emit(name, data);
     }
 
-    once(name: string,  callback: (...args: any[]) => void) {
+     /**
+     *  remove event
+     * 
+     * @param name event name
+     * @param data data event
+     */
+    public off(name: string, callback: (...args: any[]) => void) {
+        this._event_bus.off(name, callback);
+    }
+
+
+    public once(name: string,  callback: (...args: any[]) => void) {
         this._event_bus.once(name, callback);
     }
+
+
+    public registerChannel(name:string, event_name: string, callback: Function, limit: number = 0) {
+        const channel = new Channel(this, name, event_name, callback, limit);
+        this._channel.set(name, channel);
+        channel.listen();
+    }
+
+    public removeChannel(channel: Channel) {
+        channel.remove();
+        this._channel.delete(channel.name);
+    }
+
+    public getChannel(name): Channel {
+        return this._channel.get(name);
+    }
 }

tests/channel.test.ts

@@ -0,0 +1,36 @@
+import { expect } from "chai";
+import { EventBus } from "../src/EventBus";
+import { ContextChannel } from "../src/ContextChannel";
+
+
+var bus: EventBus = null;
+
+describe('Channel test', function() {
+
+    before(function () {
+        bus = new EventBus();
+    })
+
+    after(function(){
+        bus = null;
+    })
+
+    it('register', function() {
+        bus.registerChannel("test", "test", (data: ContextChannel) => {
+            expect(data.channel.name).to.eql("test");
+        });
+        var channel = bus.getChannel("test");
+        expect(channel.name).to.eql("test");
+        bus.emit("test","t");
+        bus.emit("test","t");
+        bus.emit("test","t");
+    });  
+
+    it('remove channel', function() {
+        var channel = bus.getChannel("test");
+        bus.removeChannel(channel);
+        channel = bus.getChannel("test");
+        expect(channel).to.eql(undefined);
+    });  
+
+});

tests/eventbus.test.ts

@@ -28,4 +28,13 @@ describe('EventBus test', function() {
         bus.emit("test_1", "test");
     });
 
+    it('remove event one', function() {
+        var callback =  function() {
+            expect(true).to.eql(false);
+        }
+        bus.on("test_2", callback);
+        bus.off("test_2", callback);
+        bus.emit("test_2", "test");
+    });
+
 });

tsconfig.json

@@ -1,25 +1,25 @@
 {
-    "compileOnSave": false,
-    "compilerOptions": {
-      "declaration": true,
-      "emitDecoratorMetadata": true,
-      "experimentalDecorators": true,
-      "module": "commonjs",
-      "moduleResolution": "node",
-      "noEmitOnError": true,
-      "noImplicitAny": false,
-      "noUnusedLocals": true,
-      "noUnusedParameters": true,
-      "rootDir": ".",
-      "inlineSourceMap": true,
-      "target": "es6",
-      "inlineSources": true,
-      "typeRoots": ["node_modules/@types"],
-      "pretty": true,
-      "outDir": "lib"
-    },
-    "exclude": ["node_modules"],
-    "files": [
-      "./index.ts"
-    ]
-  }
+  "compileOnSave": false,
+  "compilerOptions": {
+    "declaration": true,
+    "emitDecoratorMetadata": true,
+    "experimentalDecorators": true,
+    "module": "commonjs",
+    "moduleResolution": "node",
+    "noEmitOnError": true,
+    "noImplicitAny": false,
+    "noUnusedLocals": true,
+    "noUnusedParameters": true,
+    "rootDir": ".",
+    "inlineSourceMap": true,
+    "target": "es6",
+    "inlineSources": true,
+    "typeRoots": ["node_modules/@types"],
+    "pretty": true,
+    "outDir": "lib"
+  },
+  "exclude": ["node_modules"],
+  "files": [
+    "./index.ts"
+  ]
+}