4
\$\begingroup\$

Here is a plunker of my working demo.

I have the following code which I have created for a subscription service for data through a webworker. I was wondering if I was following functional-reactive programming best practices and patterns on the webworker. I am new to RxJS and none of our team at work has ever used it, but we really see it's potential in this demo. I am not sure how to refactor the code to avoid the use of side-effects and the like. Any tips whatsoever on the code would be good.

script.js:

var worker = new Worker('webworker.js');

var subscriberIDs= {};

var idGen = (function idGen () {
  var id = 0;
  return function () {
    return id++;
  };
})();

function renderSuggestion(suggestedUser, selector) {
    var suggestionEl = document.querySelector(selector);
    if (suggestedUser === null) {
        suggestionEl.style.visibility = 'hidden';
    } else {
        suggestionEl.style.visibility = 'visible';
        var usernameEl = suggestionEl.querySelector('.username');
        usernameEl.href = suggestedUser.html_url;
        usernameEl.textContent = suggestedUser.login;
        var imgEl = suggestionEl.querySelector('img');
        imgEl.src = "";
        imgEl.src = suggestedUser.avatar_url;
    }
}

function onError(e) {
    console.log([
      'ERROR: Line ', e.lineno, ' in ', e.filename, ': ', e.message
    ].join(''));
  }

worker.addEventListener('message', function(e) {
  if(subscriberIDs[e.data.cmd]) {
    renderSuggestion(e.data.msg, '.suggestion[data-id="' + e.data.cmd + '"]');
  }
}, false);

worker.addEventListener('error', onError, false);
$(function () {
  $(".subscribe").click(function () {
    var subID = idGen();
    subscriberIDs[subID] = true;
    $("#subscriptions").append(createSubEl(subID));
    worker.postMessage({cmd: "subscribe", msg: {subscription: "github/users", subscriberID: subID} });  
  });
  $(document).on("click", ".unsubscribe", function (e) {
    var subID = $(this).attr("data-id");
    delete subscriberIDs[subID];
    worker.postMessage({cmd: "unsubscribe", msg: {subscription: "github/users", subscriberID: subID} });  
    $(this).closest(".suggestion").remove();
  });
  $("#updateTime").on("change", function () {
    var timeVal = parseInt($(this).val(), 10);
    worker.postMessage({cmd: "updateSubTime", msg: {subscription: "github/users", time: timeVal } });
  })
});

function createSubEl (subscriberID) {
  return $(`<div class="suggestion" data-id="${subscriberID}" style="visibility:hidden;">
      <input input type="button" value="unsubscribe" class="unsubscribe pull-right btn btn-info" data-id="${subscriberID}">
      <img />
      <a href="#" target="_blank" class="username">this will not be displayed</a>
    </div>`);
}

webworker.js:

importScripts("rxjs.min.js");
importScripts("underscore.min.js");
importScripts("io.js");

var sources = {};
var subscriptions = {};

// Code goes here
self.addEventListener('message', function(e) {
  var data = e.data;
  switch (data.cmd) {
    case 'subscribe':
      var msg = data.msg;
      insertSubscription(msg.subscription, msg.subscriberID);
      break;
    case 'unsubscribe':
      var msg = data.msg;
      removeSubscription(msg.subscription, msg.subscriberID);
      break;
    case 'updateSubTime':
      var msg = data.msg;
      updateSubTime(msg.subscription, msg.time);
      break;
    default:
      self.postMessage('Unknown command: ' + data.msg);
  };
}, false);

function insertSubscription(subscription, subscriberID) {
  if(!sources[subscription]) {
    sources[subscription] = {};
    sources[subscription].subscriptions = {};
    sources[subscription].subscriptions[subscriberID] = false;
    createSource(subscription);
  }

  createSubscription(subscription, subscriberID, function (data) {
    self.postMessage({cmd: subscriberID, msg: data});
  });
}

function createSource (subscription) {
  var callInterval = 0;
  if(sources[subscription]) {
    Rx.Observable.interval(60 * 1000).startWith(0).map(function () {
      return 'https://api.github.com/rate_limit';
      /*
      example
      {
        "resources": {
          "core": {
            "limit": 5000,
            "remaining": 4999,
            "reset": 1372700873
          },
          "search": {
            "limit": 30,
            "remaining": 18,
            "reset": 1372697452
          }
        },
        "rate": {
          "limit": 5000,
          "remaining": 4999,
          "reset": 1372700873
        }
      }
      */
    }).flatMap(function (requestUrl) {
      return get(requestUrl).then(function(data) {
        var rateLimit = JSON.parse(data).resources;
        // get times in milliseconds
        var timeOfReset = rateLimit.core.reset * 1000;
        var timeNow = (new Date()).getTime();
        var numSubscriptions = Object.keys(sources[subscription].subscriptions).length;
        var callInterval =  (timeOfReset - timeNow) / (rateLimit.core.remaining + numSubscriptions);
        console.log(rateLimit);
        console.log(callInterval);
        return callInterval;
      });
    }).subscribe(function (interval) {
      callInterval = interval
    });
    if(!sources[subscription].observable || !sources[subscription].subject) {
      var randomOffset = Math.floor(Math.random()*500);
      var currentPosition = 0;
      var cacheData = [];
      sources[subscription].subject = new Rx.Subject();

      sources[subscription].observable = sources[subscription].subject
        .startWith(5000)
        .flatMapLatest(function(intvl){
          return Rx.Observable.interval(Math.max(1000, intvl, callInterval)).startWith("something");
        })
        //.bufferWithTime(1000)
        .take(10)
        .map(function() {
            return 'https://api.github.com/users?since=' + randomOffset++;
        })
        .flatMap(function (requestUrl) {
          if(cacheData[currentPosition]) {
              return Promise.resolve(cacheData[currentPosition++]);
          } else {
            currentPosition = 0;
            var promise = get(requestUrl).then(function(data) {
              cacheData = JSON.parse(data);
              var dataToSend = cacheData[currentPosition];
              currentPosition++;
              return dataToSend;
            });
            return promise;
          }
        });

      sources[subscription].subject.onNext(5000);
    }
  }
}

function updateSubTime (subscription, time) {
  if(sources[subscription] && sources[subscription].subject) {
    sources[subscription].subject.onNext(time);
  }
}

function createSubscription(subscription, subscriberID, subCallback){
  console.log("heretoo");
  if(sources[subscription] && !sources[subscription].subscriptions[subscriberID] && typeof subCallback == "function") {
    console.log("herethree");
    sources[subscription].subscriptions[subscriberID] = sources[subscription].observable.subscribe(subCallback);
  } 
}

function removeSubscription (subscription, subscriberID) {
  if(sources[subscription] && sources[subscription].subscriptions[subscriberID]) {
    sources[subscription].subscriptions[subscriberID].dispose();
    delete sources[subscription].subscriptions[subscriberID];
  }
}
\$\endgroup\$
  • \$\begingroup\$ Please state only code purpose in the title. \$\endgroup\$ – Caridorc Dec 5 '15 at 21:19
  • \$\begingroup\$ Hi! Welcome to Code Review. If there are tags you want, you can request them, and we can decide if they are fit to be a tag! \$\endgroup\$ – TheCoffeeCup Dec 5 '15 at 21:40
  • \$\begingroup\$ @TheCoffeeCup how would I go about making a tag request would I do that on meta code review with the [tag-creation] tag? \$\endgroup\$ – John Dec 5 '15 at 21:58
  • \$\begingroup\$ @John I assume so. \$\endgroup\$ – TheCoffeeCup Dec 6 '15 at 1:30
  • \$\begingroup\$ Top level var subscriptions is not used. \$\endgroup\$ – Roamer-1888 Dec 9 '15 at 15:46
1
\$\begingroup\$

OK, trying to answer the specific review objectives ...

I was wondering if I was following functional-reactive programming best practices and patterns on the webworker.

I'm not really an FRP expert so can't really comment on that aspect per se, however, I can show you how to better organise your code.

I am not sure how to refactor the code to avoid the use of side-effects and the like.

The worst feature of "webworker.js" as it stands, is the need to pass subscription to almost every function, so sources[subscription] (and its properties) can be operated on. This can be avoided by factoring the code into Constructors (classes if you like), each with a bunch of methods.

I found that the code fell very naturally into three Constructors :

  • Sources() - what was var sources = {}
  • Subscription() - what was all over the place
  • Observable() - what was createSource() but no longer creates its own subject

The three work hierarchically such that :

  • top level code (the switch-case structure) is aware of Sources()
  • Sources() is aware of Subscription()
  • Subscription() is aware of Observable()

Note: Observable() doesn't have any methods and isn't really necessary. It avoids a lot of bulk in Subscription().

webworker.js :

The result is highly readable, object-oriented code (untested) :

importScripts("rxjs.min.js");
importScripts("underscore.min.js");
importScripts("io.js");
var sources = new Sources();
self.addEventListener('message', function(e) {
    switch(data.cmd) {
        case 'subscribe':
            sources.subscribe(e.data.msg.subscription, e.data.msg.subscriberID);
        break;
        case 'unsubscribe':
            sources.unsubscribe(e.data.msg.subscription, e.data.msg.subscriberID);
        break;
        case 'updateSubTime':
            sources.updateSubTime(e.data.msg.subscription, e.data.msg.time);
        break;
        default:
            self.postMessage('Unknown command: ' + e.data.msg);
    };
}, false);

// **********************************
// *** start: Sources constructor ***
// **********************************
function Sources() {
    this.sources = {};
}
Sources.prototype.subscribe = function(subscriptionID, subscriberID) {
    if(!this.sources[subscriptionID]) {
        this.sources[subscriptionID] = (new Subscription()).updateSubTime(5000);
    }
    this.sources[subscriptionID].subscribe(subscriberID, function(data) {
        self.postMessage({cmd: subscriberID, msg: data});
    });
    return this; //for method chaining
};
Sources.prototype.unsubscribe = function(subscriptionID, subscriberID) {
    this.sources[subscriptionID].unsubscribe(subscriberID);
    return this; //for method chaining
}
Sources.prototype.updateSubTime = function(subscriptionID, time) {
    this.sources[subscriptionID].updateSubTime(time);
    return this; //for method chaining
}
// ********************************
// *** fin: Sources constructor ***
// ********************************

// ***************************************
// *** start: Subscription constructor ***
// ***************************************
function Subscription() {
    this.subject = new Rx.Subject();
    this.subscriptions = {};
    this.observable = new Observable(this.subject, this.subscriptions);
}
Subscription.prototype.subscribe = function(subscriberID, callback) {
    this.subscriptions[subscriberID] = this.observable.subscribe(callback);
    return this; //for method chaining
};
Subscription.prototype.unsubscribe = function(subscriberID) {
    if(this.subscriptions[subscriberID]) {
        this.subscriptions[subscriberID].dispose();
        delete this.subscriptions[subscriberID];
        return this; //for method chaining
    }
};
Subscription.prototype.updateSubTime = function(time) {
    this.subject.onNext(time);
    return this; //for method chaining
};
// *************************************
// *** fin: Subscription constructor ***
// *************************************

// *************************************
// *** start: Observable constructor ***
// *************************************
function Observable(subject, subscriptions) {
    var callInterval = 0;
    Rx.Observable.interval(60 * 1000).startWith(0).map(function () {
        return 'https://api.github.com/rate_limit';
    }).flatMap(function(requestUrl) {
        return get(requestUrl).then(function(data) {
            var rateLimit = JSON.parse(data).resources;
            return (rateLimit.core.reset * 1000 - Date.now()) / (rateLimit.core.remaining + Object.keys(subscriptions).length);
        });
    }).subscribe(function(interval) {
        callInterval = interval;
    });
    var randomOffset = Math.floor(Math.random() * 500);
    var currentPosition = 0;
    var cacheData = [];
    return subject
    .startWith(5000)
    .flatMapLatest(function(intvl) {
        return Rx.Observable.interval(Math.max(1000, intvl, callInterval)).startWith('something');
    })
    //.bufferWithTime(1000)
    .take(10)
    .map(function() {
        return 'https://api.github.com/users?since=' + randomOffset++;
    })
    .flatMap(function(requestUrl) {
        if(!cacheData[currentPosition]) {
            cacheData[currentPosition] = get(requestUrl).then(function(data) {
                return JSON.parse(data);
            });
        }
        var promise = cacheData[currentPosition];
        currentPosition = (currentPosition + 1) % 10;
        return promise;
    });
}
// ***********************************
// *** fin: Observable constructor ***
// ***********************************

I'm not sure about my caching strategy. From what you say in comments above, you may need to revert to what you had originally.

script.js :

"script.js" doesn't really need to be touched. The {cmd: ..., msg: ... } messaging protocol is unchanged.

I've just made a few tweeks.

// Everything can be wrapped in $(function () {...});, not just the event listeners
$(function () { 
    var worker = new Worker('webworker.js');
    var subscriberIDs = {};
    var idGen = (function idGen() {
        var id = 0;
        return function() {
            return id++;
        };
    })();
    function renderSuggestion(suggestedUser, selector) {
        if(suggestedUser) {
            $(selector).show().find('img').attr('src', suggestedUser.avatar_url).end().find('a').attr('href', suggestedUser.html_url).text(suggestedUser.login);
        } else {
            $(selector).hide();
        }
    }
    function onError(e) {
        console.log(['ERROR: Line ', e.lineno, ' in ', e.filename, ': ', e.message].join(''));
    }
    worker.addEventListener('message', function(e) {
        if(subscriberIDs[e.data.cmd]) {
            renderSuggestion(e.data.msg, '.suggestion[data-id="' + e.data.cmd + '"]');
        }
    }, false);

    worker.addEventListener('error', onError, false);

    $(".subscribe").on('click', function () {
        var subID = idGen();
        subscriberIDs[subID] = true;
        $('#subscriptions').append(createSubEl(subID));
        worker.postMessage({cmd: 'subscribe', msg: {subscription: 'github/users', subscriberID: subID} });  
    });
    $(document).on('click', '.unsubscribe', function (e) {
        var subID = $(this).closest('.suggestion').detach().data('id');
        delete subscriberIDs[subID];
        worker.postMessage({cmd: 'unsubscribe', msg: {subscription: 'github/users', subscriberID: subID} });  
    });
    $('#updateTime').on('change', function () {
        var timeVal = parseInt($(this).val(), 10);
        worker.postMessage({cmd: 'updateSubTime', msg: {subscription: 'github/users', time: timeVal } });
    })

    function createSubEl(subscriberID) {
        return $('<div class="suggestion" style="visibility:hidden;"><input type="button" value="unsubscribe" class="unsubscribe pull-right btn btn-info"><img /><a href="#" target="_blank" class="username">this will not be displayed</a></div>').data('id', subscriberID); //Note that it's only necessary to set subscriberID on the div container, and by using jQuery the ES6>>>ES5 transpiler has less to do :)
    }
});
\$\endgroup\$
  • \$\begingroup\$ John, as you know, your suggested edit was correct. I don't know why it was rejected. \$\endgroup\$ – Roamer-1888 Dec 12 '15 at 8:14

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.