lets do the subscriber here
766483
function subscriber(port) {
  var socket = zeromq.socket('sub');
  socket.identity = 'sub' + process.pid;

  // have to subscribe, otherwise nothing would show up
  // subscribe to everything
  socket.subscribe('');

  var totalMessages=0;

  // receive messages
  socket.on('message', function(data) {
    //console.log(socket.identity + ': received data ' + data.toString());
    totalMessages++;
  });

  socket.connect(port); 
  console.log('connected to ' + port);

  // lets count messages recvd
  setInterval(function() {
    console.log('messages received = ' + totalMessages);
    totalMessages=0;
  }, 1000);

}
1167052
public void _onSubscribe(Subscription subscription) {
        producer = subscription;
        ArrayList<Runnable> tmp = this.doOnSubscribe;
        this.doOnSubscribe = null;
        tmp.forEach(runnable -> runnable.run());
    }
856781
def subscribe(topic_name, topic_type, &callback)
      sub = Subscriber.new(@node_name,
                           resolve_name(topic_name),
                           topic_type,
                           callback)
      @manager.add_subscriber(sub)
      trap_signals
      sub
    end
1361410
def subscribe(self, *, create_task=True, **params):
        
        if create_task and not self.task:
            self.task = asyncio.ensure_future(self.run(**params))
        return self.channel.subscribe()
33057
def subscribe(self, feedUrl):
        
        response = self.httpPost(
            ReaderUrl.SUBSCRIPTION_EDIT_URL,
            {'ac':'subscribe', 's': feedUrl})
        # FIXME - need better return API
        if response and 'OK' in response:
            return True
        else:
            return False
74318
static <T> void subscribe(Subscriber<? super T> s, Iterator<? extends T> it) {
        if (it == null) {
            error(s, new NullPointerException("The iterator is null"));
            return;
        }

        boolean b;

        try {
            b = it.hasNext();
        } catch (Throwable e) {
            error(s, e);
            return;
        }

        if (!b) {
           complete(s);
            return;
        }

        s.onSubscribe(new IterableSubscription<T>(s, it));

    }
1067800
def subscribe(self, topic, channel):
        """"""
        self.send(nsq.subscribe(topic, channel))
700144
def subscribe_to_chat(
        self,
        subscriber: Callable[[events.ChatMessageWasReceived], None],
    ) -> None:
        
        self._chat_subscribers.append(subscriber)
1027755
def subscribe(self, subscription_channel, **observer_params):
        
        return self.get_channel(subscription_channel, **observer_params).ensure_started().observer
523273
def _subscribed(self):
        
        self.logger.debug('Subscription confirmed.')
        self.state = 'subscribed'
        for message in self.message_queue:
            self.send(message)
280541
function autoSubscribe() {
    //re-add pending subscribers
    if (_pendingSubscribers.length > 0) {
      async.each(_pendingSubscribers, function(sub, callback) {
        debug('Add pending subscriber', sub);
        self.subscribeToTopic(sub.exchange, sub.queue, sub.filter, sub.subscriber, sub.opts, function(err) {
          if (err) {
            debug('Failed to add subscriber, keep it', sub);
          } else {
            //done, remove the item from the pending subscribers
            var idx = _pendingSubscribers.indexOf(sub);
            _pendingSubscribers.splice(idx, 1);
            debug('pending subsriber added, now there are ' + _pendingSubscribers.length + ' left');
          }
          return callback();
        });
      }, function() {
        debug('pending subscribers are added');
      });
    }
  }
210137
def subscribe(self, request, *args, **kwargs):
        
        self.object = self.get_object()
        self.object.subscribers.add(request.user)
        messages.success(self.request, self.success_message)
        return HttpResponseRedirect(self.get_success_url())
969181
@Override
    public void subscribe(final Subscriber<? super T> s) {

       transformerStream().forEach(v->v.subscribe(s));

    }
1077821
def present_subscribers
      subs = {}
      subscribers.each do |k, v|
        unless v.empty?
          subs[k] = v
        end
      end
      subs
    end
187143
function execute(req, res) {
  this.state.pubsub.punsubscribe(req.conn, req.args);
}
1202346
def subscribe(caller = nil, &block)
      if block_given?
        notify_when = block.call(state)
        @listeners << { obj: block.binding.receiver, notify_when: notify_when }
      else
        @listeners << { obj: caller }
      end
    end
502953
function Subscriber(opts) {
  opts = opts || {};

  this._channel = opts.channel || 'default';
  this._subscribers = [];

  /* for mocking */
  var addEventListener = opts.addEventListener || window.addEventListener;
  var attachEvent = opts.attachEvent || window.attachEvent;

  var _this = this;
  function onMessage(e) {
    var event = parse(e.data);
    if (event.channel === _this._channel) {
      for (var i = 0, len = _this._subscribers.length; i < len; i++) {
        _this._subscribers[i].apply({}, event.args);
      }
    }
  }

  if (addEventListener) {
    addEventListener('message', onMessage, false);
  } else {
    attachEvent('onmessage', onMessage, false);
  }
}