Skip to content

Instantly share code, notes, and snippets.

@Thermatix
Last active December 25, 2016 19:12
Show Gist options
  • Save Thermatix/a82819dffa914a421ea4039b68a6ceb3 to your computer and use it in GitHub Desktop.
Save Thermatix/a82819dffa914a421ea4039b68a6ceb3 to your computer and use it in GitHub Desktop.
function ReplayableEventStore (settings) {
var self = this;
var config = {
warnings: def_value(settings.warns,true),
replayPre: def_value(settings.replayPre,true)
};
self.preSubscribeEvents = {};
self.channelIndex = {};
self.store = [];
self.end = function (){
self.store.size - 1
}
self.channelHandlers = {};
self.flush = function () {
self.store = [];
for (var channel in self.channelIndex) {
self.channelIndex[channel] = [];
}
};
self.replay = function (selector) {
selector = def_value(selector,null);
if(self.store.size == 0) {
warn("No events in the event store to replay")
}else{
if (typeof selector == "string") {
replay_this(selector, 0, self.channelIndex[selector].size, function (i,ch) {
self.channelIndex[ch][i];
})
} else if (typeof selector == "array") {
replay_this(self.store, selector[0], selector[1], function (i,ch) {
self.store[i];
});
} else {
end = selector || self.store.size
replay_this(self.store, 0, end, function (i,ch) {
self.store[i];
});
}
}
};
function replay_this (channel,begin,end) {
is_array = channel.constructor === Array
for(var i = begin; i < end; i ++) {
ChannelCallBack((ch = is_array ? channel[i].channel : chanel), self.store[i]);
};
};
self.subscribe = function (channel,callback) {
if(channel){
if(callback.constructor === Array){
id = batch_sub(channel,callback);
}else{
id = sub(channel,callback);
}
if(config.replayPre){
if(self.preSubscribeEvents[channel]){
self.replay(channel);
preSubscribeEvents[channel] = false;
}
}
return id
} else {
warn("No channel Given", true);
}
};
self.unsubscribe = function (channel,id) {
if(typeof id == "array"){
for(var i = 0; i < id.size; i++){
destroySub(channel,id[i]);
}
}else{
destroySub(channel,id);
}
}
self.emit = function (channel,data) {
if(channel){
obj = {
channel: channel,
msg: def_value(data,null)
};
indexPos = self.store.size;
self.store.push(obj);
setInitial("channelIndex",channel);
self.channelIndex[channel].push(indexPos);
ChannelCallBack(channel,obj);
}else{
warn("No channel Given",true);
}
};
self.dump = function () {
return {
store: self.store,
channelIndex: self.channelIndex
};
};
self.load = function (dump){
self.store = dump.store;
self.channelIndex = dump.channelIndex || {};
}
function batch_sub (channel,callbacks) {
m_subs = []
for (var i = 0; i < callbacks.size; i++) {
m_subs.push(sub(channel,callbacks[1]));
}
return m_subs;
};
function sub (channel,callback) {
//batch = def_value(batch,false);
setInitial("channelHandlers",channel);
if(callback){
var id = self.channelHandlers[channel].size;
self.channelHandlers[channel].push(callback);
return id;
}else{
warning("No Callback Given",true);
}
};
function setInitial (prop,ident) {
if(!self[prop][ident]){
self[prop][ident] = [];
}
};
function destroySub (channel,id){
if(channelHandlers[channel]){
if(channelHandlers[channel][id]){
channelHandlers[channel][id] = null;
}else{
warning("No matching handler for given ID");
}
}else{
warning("No handlers available for given channel");
}
}
function ChannelCallBack (channel,data) {
if (self.channelHandlers[channel]) {
for (var i = 0; i < self.channelHandlers[channel].length; i++) {
if(self.channelHandlers[channel][i]){
self.channelHandlers[channel][i](data);
}
}
}else{
if(config.replayPre){
setInitial("preSubscribeEvents",channel);
self.preSubscribeEvents[channel] = true;
}else{
warning("No handlers for given channel: " + channel);
}
}
};
function def_value (varb,def){
return typeof varb !== "undefined" ? varb : def;
};
function warning (message,throwthis=false) {
if(throwthis){
throw new Error(message);
}else{
if(config.warnings){
console.warn(message);
return false;
}
}
};
};
@Thermatix
Copy link
Author

Thermatix commented Jul 11, 2016

note: replay mechanism doesn't work properly,
Can be used like this:

// create event store
//config options
// warns: turn warnings off or on, default true
//replayPre: turns on replaying of events emmited before there is a subscriber to consume those events, default true,
// when doing batch subscribers, the events won't replay untill all the subscribers have been added.
var events = ReplayableEventStore.new({warns: false});

// add subscribers
events.subscribe('SomeEvent', function(data){
  // data is {channel: 'SomeEvent',msg: {key: 'value'}}
  // some code in response to this event
});

var subscriberId = events.subscribe('SomeEvent', function(data){
  // some code in response to this event
});

//add multiple subscribers
var MultiSubs = events.subscribe("SomeEvent",[
  function(data){
  // some code in response to this event
  },
  function(data){
    // some code in response to this event
  }
]

// any subscribers will be executed in the order they were added
events.emit('SomeEvent',{key: 'value'});
events.emit('SomeEvent',{key: 'value1'});
events.emit('SomeEvent',{key: 'value2'});

// remove subscriber
events.unSubscribe('SomeEvent', subscriberId);

//remove multiple subscribers
events.unSubscribe('SomeEvent',MultiSubs);

// replay events
// All events for a channel
events.replay('SomeEvent');

// All events in a range
events.replay([1,2]);

// All events up to a point
events.replay(2);

// All events from a point
events.replay([1,events.end]);

// All events ever
events.replay();

// Destroy all events
events.flush();


// note, this won't dump subscribers, just the events and the channel index
// Dump events
var eventsDump = events.dump();

// Load events from Dump
events.load(eventsDump);

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment