HyperEmitter   

HyperEmitter is a horizontally scalable and persistent EventEmitter powered by a Merkle DAG. (Yes, it's like a blockchain). In other contexts, this concept is also called an EventStore. HyperEmitter uses protocol-buffers, specifically mafintosh's implementation, for handling message schemas, although custom codecs are also supported.

This module is highly experimental, possibly under-perfoming, and may have bugs, use with caution. On the other end, if the thought of a persistent, horizontally scaleable EventStore gets you excite please jump in wherever you feel you can help!

Installation

To install the module for use locally in your project use:

npm install hyperemitter --save

To install the companion CLI tool, you will need to install globally:

npm install hyperemitter -g

Example

The example below can be found and ran from the examples folder; it demonstrates how to connect two HyperEmitters together and how they both receive all messages sent.

'use strict'

var fs = require('fs')
var path = require('path')

// The emitter itself as well as an in memory
// leveldb based store, any leveldb store will do.
var HyperEmitter = require('../hyperemitter')
var buildDB = require('memdb')

// use the example-schema.proto as the message schema.
var schema = fs.readFileSync(path.join('.', 'example-schema.proto'))

// two emitters will be used for this example, notice each
// maintains it's own leveldb store and share the same schema.
var emitterOne = new HyperEmitter(buildDB('a'), schema)
var emitterTwo = new HyperEmitter(buildDB('b'), schema)

// listen on port 9001, ensure no connection error.
emitterOne.listen(9901, function (err) {
  if (err) { return }

  // connect to the first emitter.
  emitterTwo.connect(9901, '127.0.0.1', function (err) {
    if (err) { return }
  })

  // basic message type
  var userAddedMsg = {
    id: 1,
    username: 'user'
  }

  // basic message type
  var userRemovedMsg = {
    id: 1
  }

  // Messages sent on either emitter will be handled.
  emitterOne.on('userRemoved', function (msg) {
    console.log('userRemoved: ', msg)
  })

  // Messages sent on either emitter will be handled.
  emitterTwo.on('userAdded', function (msg) {
    console.log('userAdded: ', msg)
  })

  // We send each message across the opposite emitter.
  emitterOne.emit('userAdded', userAddedMsg)
  emitterTwo.emit('userRemoved', userRemovedMsg)

  function complete () {
    emitterOne.close()
    emitterTwo.close()
  }

  // we will wait for 500ms to see if more than one
  // message is delivered to the subscribers above.
  setTimeout(complete, 500)
})

Using The CLI Tool

HyperEmitter comes with a nice CLI to work with other remote HyperEmitter's. Also, it doubles as really useful tool for debugging. The sample below uses the schema in test/fixture/.

hypem --db db --port 1234 fixture/messages.proto

On pressing enter, you should see output similar to the following:

listening on 1234 127.0.0.1
EventPeer  { id: '123', addresses: [ { ip: 'localhost', port: 1234 } ] }

From here you can then use the provided REPL to work with HyperEmitter:

hyper.emit('Hello', { from: 'Matteo' })

And receive the following response:

Hello { from: 'Matteo', message: '' }

To connect another hyperEmitter in another shell:

hypem --db db2 --target-port 1234 fixture/messages.proto

Which will output the following:

connected to localhost 1234
EventPeer { id: '123', addresses: [ { ip: 'localhost', port: 1234 } ] }
Hello { from: 'Matteo', message: '' }

As you can see the events are synced up!

To export the data in newline delimited json, use:

hypem --no-repl --db db4 --target-port 1234 fixture/messages.proto

Which will produce output like so:

{"name":"EventPeer","payload":{"id":"123","addresses":[{"ip":"localhost","port":1234}]}}
{"name":"Hello","payload":{"from":"Matteo","message":""}}

The cli tool also works as a input stream, following the UNIX philosophy. If you close a REPL or a ndjson stream, the next time it will start where it stopped. If you have a stream, you can start back from the beginning using the --from-scratch flag.

API Reference

This page contains a list of public API's exposed by the HyperEmitter module as well as a brief description of their use. For additional samples please checkout out our examples folder.

• HyperEmitter
• .messages
• .registerCodec()
• .emit()
• .on()
• .removeListener()
• .connect()
• .listen()
• .stream()
• .close()

HyperEmitter(db, schema, [opts])

HyperEmitter is both the name of the module and of the function to be required. The function is actually a class or a class factory depending on how it is used. The code below shows both valid ways to get a new instance of HyperEmitter for use:

var emitterOne = Hyperemitter(someDb, someSchema)
var emitterTwo = new HyperEmitter(someDb, someSchema)

• db

The db argument accepts a levelup instance, which in turn is powered by leveldb. We recommend level for persistent storage and memdb if you require an in memory store.

• schema

An string or stream that represents all of the messages to support in a given instance of HyperEmitter.

• opts

An optional object that can be provided to configure the created instance. All available options are listed below.

• reconnectTimeout: The timeout that this instance will wait before reconnecting to peers.

---

.messages

Message types are stored in the .messages field for each instance created. This field is populated with a normalized object based on the schema provided. Each property represents a single message, indvidual messages, as well as their encoder and decoder can be accessed from here.

var emitter = HyperEmitter(someDb, someSchema)

// the object containing each message as a property
var messages = emitter.messages

// access an individual message by it's name
var message = messages[messageName]

// access a given message's encode / decode functions
var encoder = message.encode
var decode = message.decode

---

.emit(event, message, [callback])

Messages can be emitted from HyperEmitter using the .emit() method. This method takes the name of the event to be emitted and validates message against the schema before sending it off to any listening subscribers, in parallel. Once complete the callback function is called, if present.

• event

The name of one of the message definitions from the provided schema.

• message

Any object who's shape matches the named event. It's keys are validated against the schema.

• callback

An optional function that will be called once the emitted message has been added to the log.

---

.registerCodec(name, codec | codecs)

Custom codecs can be registered as long as they have both an encode and decode method. Codecs are keyed by message name. For ease of use registration params can be provided as args (name, codec), an object, or an array of {name: '', codec: obj}. Only once codec can be registered against one message at any given time.

• name

The name of the message message this codec handles.

• codec

Any object which has an encode and decode method.

• codecs

An object or array which represents a collection of codecs and their names.

---

.on(event, callback(message[, done]))

Subscribes to and provides a function to be called each time and event is raised.

• event

The name of the event being subscribed to.

• callback

The function to be called when a new event is raised. The message arg holds the message emitted. The done arg can be used for letting the emitter know when the function has completed the handling of the event.

---

.removeListener(event, callback(message[, done]))

Removes the listener who matches the one provided. This method does not work with anonymous functions, only a function with a prior reference can be removed.

• event

The name of the event the listener is subscribed to.

• callback

The reference of the function originally used in the .on call.

---

.connect(port, host[, done])

Connects this HyperEmitter with another one which we call a peer. Peers can exist on other machines, HyperEmitter will communicate over TCP using the host and port provided. An optional function can be provided that will be called once the connection has been made.

• port

The port of the machine the peer to connect to resides on

• host

The host of the machine the peer to connect to resides on.

• done

A function to be called when connected.

---

.listen(port[, host[, done]])

Listen on a given port/host combination. An EventPeer event will be emitter.

• port

The port to listen on.

• host

The host to listen on.

• done

An optional function to be called one listening has begun.

---

.stream([opts])

A Duplex stream to emit and receive events.

• opts

An optional object of settings.

• from: The point in the stream to return events since. supports 'beginning'

---

.close(callback)

Close a given HyperEmitter. After, all emit will return an error.

• callback

An optional function to be called when close has completed.

---

Contributing

HyperEmitter is a mad science project and like all mad science project it requires mad scientists, the more the merrier! If you feel you can help in any way, be it with examples, extra testing, or new features please be our guest. See our Contribution Guide for information on obtaining the source and an overview of the tooling used.

License

Copyright Matteo Collina 2015, Licensed under ISC