[END pubsub_listen_ordered_messages]
439084
async function listenForErrors(subscriptionName, timeout) {
  // [START pubsub_subscriber_error_listener]
  // Imports the Google Cloud client library
  const {PubSub} = require('@google-cloud/pubsub');

  // Creates a client
  const pubsub = new PubSub();

  /**
   * TODO(developer): Uncomment the following lines to run the sample.
   */
  // const subscriptionName = 'my-sub';
  // const timeout = 60;

  // References an existing subscription
  const subscription = pubsub.subscription(subscriptionName);

  // Create an event handler to handle messages
  const messageHandler = function(message) {
    // Do something with the message
    console.log(`Message: ${message}`);

    // "Ack" (acknowledge receipt of) the message
    message.ack();
  };

  // Create an event handler to handle errors
  const errorHandler = function(error) {
    // Do something with the error
    console.error(`ERROR: ${error}`);
  };

  // Listen for new messages/errors until timeout is hit
  subscription.on(`message`, messageHandler);
  subscription.on(`error`, errorHandler);

  setTimeout(() => {
    subscription.removeListener(`message`, messageHandler);
    subscription.removeListener(`error`, errorHandler);
  }, timeout * 1000);
  // [END pubsub_subscriber_error_listener]
}
111995
def messages(self, channel='sse'):
        
        pubsub = self.redis.pubsub()
        pubsub.subscribe(channel)
        for pubsub_message in pubsub.listen():
            if pubsub_message['type'] == 'message':
                msg_dict = json.loads(pubsub_message['data'])
                yield Message(**msg_dict)
1033716
function onPeerMessage(peer,msg)
    {
        scope.dispatchEvent("peerMessage",{message:msg,id:peer.id,rtc:peer.rtcId,array:false,fallback:false});
    }
243173
protected void finishCompoundMessage ()
    {
        if (--_compoundDepth == 0) {
            CompoundDownstreamMessage downstream = _compound;
            _compound = null;
            if (!downstream.msgs.isEmpty()) {
                postMessage(downstream, null);
            }
        }
    }
835346
function readNewMessages(state) {
  // channel already closed
  if (state.closed) return Promise.resolve(); // if no one is listening, we do not need to scan for new messages

  if (!state.messagesCallback) return Promise.resolve();
  return getMessagesHigherThen(state.db, state.lastCursorId).then(function (newerMessages) {
    var useMessages = newerMessages.map(function (msgObj) {
      if (msgObj.id > state.lastCursorId) {
        state.lastCursorId = msgObj.id;
      }

      return msgObj;
    }).filter(function (msgObj) {
      return _filterMessage(msgObj, state);
    }).sort(function (msgObjA, msgObjB) {
      return msgObjA.time - msgObjB.time;
    }); // sort by time

    useMessages.forEach(function (msgObj) {
      if (state.messagesCallback) {
        state.eMIs.add(msgObj.id);
        state.messagesCallback(msgObj.data);
      }
    });
    return Promise.resolve();
  });
}
1261740
async def __run_listen(self):
        
        while True:
            message = await ZMQUtils.recv(self.__backend_socket)
            await self.__handle_backend_message(message)
887414
@Override
    public void doMessageReceivedListeners(final Object message) {
        runManagementTask(new Runnable() {
            @Override
            public void run() {
                try {
                    List<SessionManagementListener> sessionListeners = getManagementListeners();
                    for (final SessionManagementListener listener : sessionListeners) {
                        listener.doMessageReceived(SessionManagementBeanImpl.this, message);
                    }

                    markChanged();
                } catch (Exception ex) {
                    logger.warn("Error during doMessageReceived session listener notifications:", ex);
                }
            }
        });
    }
357196
function notifyHandler(data) {
        const notification = Notification.parse(data);
        return evalBlacklist(notification)
            .then(() => {
                forum.emit('log', `Notification ${notification.id}: ${notification.label} received`);
    
                const ids = {
                    post: notification.postId,
                    topic: notification.topicId,
                    user: notification.userId,
                    pm: -1,
                    chat: -1
                };
                return notification.getText()
                    .then((postData) => forum.Commands.get(ids,
                        postData, (content) => forum.Post.reply(notification.topicId, notification.postId, content)))
                    .then((commands) => {
                        if (commands.commands.length === 0) {
                            debug(`Emitting events: 'notification' and 'notification:${notification.type}'`);
                            forum.emit(`notification:${notification.type}`, notification);
                            forum.emit('notification', notification);
                        }
                        return commands;
                    })
                    .then((commands) => commands.execute());
            }).catch((err) => {
                if (err === 'Ignoring notification') {
                //We do not process the notification, but we can continue with life
                    return Promise.resolve();
                }
                throw err;
            });
    }
1293106
def _pushMessages(self):
        
        # This continues to get itself called until no msgs are left in list.
        self.showStatus('')
        if len(self._statusMsgsToShow) > 0:
            self.top.after(200, self._pushMessages)
1046182
def preview
      notifications = Correspondent::Notification.for_subscriber(params[:subscriber_type], params[:subscriber_id])

      if stale?(notifications)
        render(
          json: {
            count: notifications.count,
            notification: notifications.limit(1).first
          }
        )
      end
    end
1052195
function finishMessage() {
    if (parser.tag_ < Parser.PADDING_TAG_) {
      var message = {};
      message[parser.tag_] = parser.messageBuffer_;
      parser.result_.push(message);
    }
    parser.state_ = Parser.State_.INIT;
  }
886821
function subscribe() {
  openSTNotification.subscribeEvent.rabbit(
    ['#'],
    {queue: 'openst_platform'},
    function (msgContent) {
      logger.debug('[RECEIVED]', msgContent, '\n');
    }
  ).catch(function (err) {
    logger.error(err);
  });
}
540027
function Pubsub(bus, name) {
  events.EventEmitter.call(this);
  this.bus = bus;
  this.type = 'pubsub';
  this.id = "bus:pubsub:" + name;
  this.logger = bus.logger.withTag(this.id);
  this.name = name;
}
121923
def event_subscription(event)
      ActiveSupport::Notifications.subscribe(event[:id]) do |*, event_id, data|
        filtered_data = data.slice(*event[:allowed_data])
        filtered_data[:event_name] = event[:id]
        filtered_data[:event_id] = event_id
        if event[:id] == "sql.active_record"
          binds = data[:binds].each_with_object({}) { |bind, output| output[bind.name] = '?' if defined?(bind.name) }
          filtered_data[:binds] = JSON.dump(binds) unless binds.empty?
        end
        Bugsnag.leave_breadcrumb(
          event[:message],
          filtered_data,
          event[:type],
          :auto
        )
      end
    end
555724
def recent_messages(limit = nil, &callback)
      connection.http(:get, "/room/#{id}/recent.json", (limit ? { :limit => limit } : nil)) do |data, http|
        callback.call(data[:messages].map { |msg| Firering::Message.instantiate(connection, msg) }) if callback
      end
    end
591591
def listen():
        """"""
        msg = MSG()
        ctypes.windll.user32.GetMessageA(ctypes.byref(msg), 0, 0, 0)
743548
function(postmessage) {
			var b = gui.Broadcast.parse(postmessage);
			if(b.$contextids.indexOf(gui.$contextid) === -1) {
				gui.Broadcast._dispatch(b);
			}
		}