<commit_before>/*
Just supply qlobber-fsq obj and stream. Multiplex (using bpmux) off that.
No user/auth stuff here.

Messages (published and received) on separate streams. Use handshake data
to describe the messages.

Subscribe, unsubscribe on a control channel. Use frame-stream

Also allow subscribe, unsubscribe from not in the stream.

So take:

- fsq object; up to caller to specify dedup - without, client may receive
  messages multiple times
- stream
- options:
    - handshake data for the control channel (e.g. up to caller if knows single
      is disabled to send that info if the client can deal with it)
    - whether to use fastest-writable
    - options for bpmux, frame-stream, fastest-writable

Exposes:

- subscribe (doesn't take handler)
- unsubscribe (doesn't take handler) - no topic means unsubscribe all
- publish
- events:
  - subscribe_requested (if no handler then calls subscribe)
  - unsubscribe_requested (if no handler then calls unsubscribe)
  - publish_requested (if no handler then calls publish)
  - handshake (on control channel only)

Lifecycle:

- Don't replicate lifecycle events of fsq obj or stream
- Only care about cleaning up our own state. 
  - On stream end or finish, unsubscribe all

Misc:

- Need to remember all active subscriptions so can remove when unsubscribe all
  Might as well use this to stop multiple subscriptions for same subject
  (although fsq dedup also takes care of that)

- How use fastest-writable? We'll have multiple mqlobber objects each with
  their own handler, so the data will be written at the speed of the slowest.
  We should put a fastest-writable onto the stream as a property, and pipe the
  stream onto it (first time we create the fw). Add our stream as peer on fw.
  (Alternative is to use info to store fw).
    - Make this optional

- We want to use the filter handler function to prevent the message being
  delivered if all streams are full.
    - How detect stream is full? write empty data
    - But the filter handler is passed when the fsq is constructed.
      This is optional behaviour, expose a filter function which caller
      can use when constructing fsq
*/

var EventEmitter = require('events').EventEmitter,
    BPMux = require('bpmux').BPMux,
    frame = require('frame-stream'),
    FastestWritable = require('fastest-writable').FastestWritable,
    util = require('util');

function MQlobber(fsq, stream, options)
{
    EventEmitter.call(this);

    options = options || {};

    this._fsq = fsq;
    this._mux = new BPMux(stream, options);
    this._subs = new Set();
    this._fastest_writable = options.fastest_writable;

    var ths = this;

    this._mux.once('handshake', function (duplex, client_data, delay)
    {
        ths._control = duplex;
        ths.emit('handshake', client_data);
        if (options.handshake_data !== undefined)
        {
            delay()(options.handshake_data);
        }

        // register for more handshake events (publish)
        // read from control for sub/unsub
        //   - frame-stream
    });
}

util.inherits(MQlobber, EventEmitter);

<commit_msg>Process messages on control stream<commit_after>/*
Just supply qlobber-fsq obj and stream. Multiplex (using bpmux) off that.
No user/auth stuff here.

Messages (published and received) on separate streams. Use handshake data
to describe the messages.

Subscribe, unsubscribe on a control channel. Use frame-stream

Also allow subscribe, unsubscribe from not in the stream.

So take:

- fsq object; up to caller to specify dedup - without, client may receive
  messages multiple times
- stream
- options:
    - handshake data for the control channel (e.g. up to caller if knows single
      is disabled to send that info if the client can deal with it)
    - whether to use fastest-writable
    - options for bpmux, frame-stream, fastest-writable

Exposes:

- subscribe (doesn't take handler)
- unsubscribe (doesn't take handler) - no topic means unsubscribe all
- publish
- events:
  - subscribe_requested (if no handler then calls subscribe)
  - unsubscribe_requested (if no handler then calls unsubscribe)
  - publish_requested (if no handler then calls publish)
  - handshake (on control channel only)

Lifecycle:

- Don't replicate lifecycle events of fsq obj or stream
- Only care about cleaning up our own state. 
  - On stream end or finish, unsubscribe all

Misc:

- Need to remember all active subscriptions so can remove when unsubscribe all
  Might as well use this to stop multiple subscriptions for same subject
  (although fsq dedup also takes care of that)

- How use fastest-writable? We'll have multiple mqlobber objects each with
  their own handler, so the data will be written at the speed of the slowest.
  We should put a fastest-writable onto the stream as a property, and pipe the
  stream onto it (first time we create the fw). Add our stream as peer on fw.
  (Alternative is to use info to store fw).
    - Make this optional

- We want to use the filter handler function to prevent the message being
  delivered if all streams are full.
    - How detect stream is full? write empty data
    - But the filter handler is passed when the fsq is constructed.
      This is optional behaviour, expose a filter function which caller
      can use when constructing fsq
*/

var EventEmitter = require('events').EventEmitter,
    BPMux = require('bpmux').BPMux,
    frame = require('frame-stream'),
    FastestWritable = require('fastest-writable').FastestWritable,
    util = require('util'),
    TYPE_SUBSCRIBE = 0,
    TYPE_UNSUBSCRIBE = 1;

function MQlobber(fsq, stream, options)
{
    EventEmitter.call(this);

    options = options || {};

    this._fsq = fsq;
    this._mux = new BPMux(stream, options);
    this._subs = new Set();
    this._fastest_writable = options.fastest_writable;

    var ths = this;

    this._mux.on('error', function (err)
    {
        ths.emit('error', err);
    });

    this._mux.once('handshake', function (duplex, hdata, delay)
    {
        ths._control = frame.decode(options);
        duplux.pipe(this._control);

        ths._control.on('readable', function ()
        {
            var data = this.read();

            if (data.length === 0)
            {
                return ths.emit('error', new Error('empty buffer'));
            }

            var type = buf.readUInt8(0, true);

            switch (type)
            {
                case TYPE_SUBSCRIBE:
                    var topic = buf.toString('utf8', 1);
                    if (!ths.emit('subscribe_requested', topic))
                    {
                        ths.subscribe(topic);
                    }
                    break;

                case TYPE_UNSUBSCRIBE:
                    var topic = buf.toString('utf8', 1);
                    if (!ths.emit('unsubscribe_requested', topic))
                    {
                        ths.unsubscribe(topic);
                    }
                    break;

                default:
                    ths.emit('error', new Error('unknown type:' + type)
                    break;
            }
        });

        this.on('handshake', function (duplex, hdata)
        {
            if (hdata.length === 0)
            {
                return ths.emit('error', new Error('empty buffer'));
            }

            var options = { single: hdata.readUInt8(0, true) },
                topic = hdata.toString('utf8', 1);

            if (!ths.emit('publish_requested', topic, duplex, options))
            {
                duplex.pipe(ths.publish(topic, options));
            }
        });

        ths.emit('handshake', hdata, delay);
    });
}

util.inherits(MQlobber, EventEmitter);

MQlobber.prototype.subscribe = function (topic, cb)
{

};

MQlobber.prototype.unsubscribe = function (topic, cb)
{

};

MQlobber.prototype.publish = function (topic, payload, options)
{

};

// have a single handler for messages
// re-emit error on fastest-writable
// filter function

{'fraction_non_alphanumeric': 0.08458646616541353, 'fraction_numerical': 0.0017543859649122807, 'mean_word_length': 2.545535317636606, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3243332', 'n_tokens_mistral': 2283, 'n_tokens_neox': 2185, 'n_words': 857}