This is dumb function, it just passes everything it gets into the
        message stream.  Something else in the stream should be responsible
        for asynchronously figuring out what to do with all these messages.
335491
def comet_view(self, request):
        
        request_id = self.id_for_request(request)
        if not request_id:
            request.write(HttpResponse(403).as_bytes())
            request.finish()
            return
        
        data = {
            'headers': request.headers,
            'arguments': request.arguments,
            'remote_ip': request.remote_ip,
            'request_id': request_id,
        }
        message_kind = 'comet-%s' % (request.method,)
        if request.method == 'POST':
            data['body'] = simplejson.loads(request.body)
            request.write(HttpResponse(201).as_bytes())
            request.finish()
        else:
            self.pending_requests[request_id].append(request)
        self.publish(Message(message_kind, datetime.now(), data))
211737
function OutgoingMessage() {
  // * This is basically a read-only wrapper for the [Stream](protocol/stream.html) class.
  Writable.call(this);

  this._headers = {};
  this._trailers = undefined;
  this.headersSent = false;
  this.finished = false;

  this.on('finish', this._finish);
}
1119931
function(msg, node, RED, io, ioNs, log) {
        node.rcvMsgCount++
        log.verbose(`[${node.url}] msg received via FLOW. ${node.rcvMsgCount} messages received`, msg)

        // If the input msg is a uibuilder control msg, then drop it to prevent loops
        if ( msg.hasOwnProperty('uibuilderCtrl') ) return null

        //setNodeStatus({fill: 'yellow', shape: 'dot', text: 'Message Received #' + node.rcvMsgCount}, node)

        // Remove script/style content if admin settings don't allow
        if ( node.allowScripts !== true ) {
            if ( msg.hasOwnProperty('script') ) delete msg.script
        }
        if ( node.allowStyles !== true ) {
            if ( msg.hasOwnProperty('style') ) delete msg.style
        }

        // pass the complete msg object to the uibuilder client
        // TODO: This should have some safety validation on it!
        if (msg._socketId) {
            log.debug(`[${node.url}] msg sent on to client ${msg._socketId}. Channel: ${node.ioChannels.server}`, msg)
            ioNs.to(msg._socketId).emit(node.ioChannels.server, msg)
        } else {
            log.debug(`[${node.url}] msg sent on to ALL clients. Channel: ${node.ioChannels.server}`, msg)
            ioNs.emit(node.ioChannels.server, msg)
        }

        if (node.fwdInMessages) {
            // Send on the input msg to output
            node.send(msg)
            log.debug(`[${node.url}] msg passed downstream to next node`, msg)
        }

        return msg
    }
1358013
function readMessageInChunksFromOutbox(message_hash, len, handleMessage){
	var CHUNK_LEN = 1000000;
	var start = 1;
	var message = '';
	function readChunk(){
		db.query("SELECT SUBSTR(message, ?, ?) AS chunk FROM outbox WHERE message_hash=?", [start, CHUNK_LEN, message_hash], function(rows){
			if (rows.length === 0)
				return handleMessage();
			if (rows.length > 1)
				throw Error(rows.length+' msgs by hash in outbox, start='+start+', length='+len);
			message += rows[0].chunk;
			start += CHUNK_LEN;
			(start > len) ? handleMessage(message) : readChunk();
		});
	}
	readChunk();
}
972741
function (data) {
        var msg = messages.unwrap(data);
        if (!msg) {
            logger.error("routeMessage got a NULL coap message ", { coreID: this.getHexCoreID() });
            return;
        }

        this._lastMessageTime = new Date();

        //should be adequate
        var msgCode = msg.getCode();
        if ((msgCode > Message.Code.EMPTY) && (msgCode <= Message.Code.DELETE)) {
            //probably a request
            msg._type = messages.getRequestType(msg);
        }

        if (!msg._type) {
            msg._type = this.getResponseType(msg.getTokenString());
        }

        //console.log("core got message of type " + msg._type + " with token " + msg.getTokenString() + " " + messages.getRequestType(msg));

        if (msg.isAcknowledgement()) {
            if (!msg._type) {
                //no type, can't route it.
                msg._type = 'PingAck';
            }
            this.emit(('msg_' + msg._type).toLowerCase(), msg);
            return;
        }


        var nextPeerCounter = ++this.recvCounter;
        if (nextPeerCounter > 65535) {
            //TODO: clean me up! (I need settings, and maybe belong elsewhere)
            this.recvCounter = nextPeerCounter = 0;
        }

        if (msg.isEmpty() && msg.isConfirmable()) {
            this._lastCorePing = new Date();
            //var delta = (this._lastCorePing - this._connStartTime) / 1000.0;
            //logger.log("core ping @ ", delta, " seconds ", { coreID: this.getHexCoreID() });
            this.sendReply("PingAck", msg.getId());
            return;
        }

        if (!msg || (msg.getId() != nextPeerCounter)) {
            logger.log("got counter ", msg.getId(), " expecting ", nextPeerCounter, { coreID: this.getHexCoreID() });

            if (msg._type == "Ignored") {
                //don't ignore an ignore...
                this.disconnect("Got an Ignore");
                return;
            }

            //this.sendMessage("Ignored", null, {}, null, null);
            this.disconnect("Bad Counter");
            return;
        }

        this.emit(('msg_' + msg._type).toLowerCase(), msg);
    }
1332680
protected void shardMessageHandler(DirectBuffer buffer, int offset, int length, Header header) {
        /**
         * All incoming messages here are supposed to be unicast messages.
         */
        // TODO: implement fragmentation handler here PROBABLY. Or forbid messages > MTU?
        //log.info("shardMessageHandler message request incoming...");
        byte[] data = new byte[length];
        buffer.getBytes(offset, data);

        VoidMessage message = VoidMessage.fromBytes(data);
        if (message.getMessageType() == 7) {
            // if that's vector request message - it's special case, we don't send it to other shards yet
            //log.info("Shortcut for vector request");
            messages.add(message);
        } else {
            // and send it away to other Shards
            publicationForShards.offer(buffer, offset, length);
        }
    }
532519
@Override
    public IQueueMessage<ID, DATA> take() throws QueueException.EphemeralIsFull {
        try {
            MessageConsumer consumer = getMessageConsumer();
            synchronized (consumer) {
                // Message message = consumer.receiveNoWait();
                Message message = consumer.receive(1000);
                if (message instanceof BytesMessage) {
                    BytesMessage msg = (BytesMessage) message;
                    byte[] buff = new byte[(int) msg.getBodyLength()];
                    msg.readBytes(buff);
                    return deserialize(buff);
                } else {
                    return null;
                }
            }
        } catch (Exception e) {
            throw e instanceof QueueException ? (QueueException) e : new QueueException(e);
        }
    }
1158600
function(stream, callback) {
                    var content = '';
                    stream.on('data', function(chunk) {
                      content += chunk;
                    });
                    stream.on('end', function() {
                      callback(content);
                    });
                  }
366103
function addMessage(from, target, text, time) {
	var name = (target == '*' ? 'all' : target);
	if(text === null) return;
	if(time == null) {
		// if the time is null or undefined, use the current time.
		time = new Date();
	} else if((time instanceof Date) === false) {
		// if it's a timestamp, interpret it
		time = new Date(time);
	}
	//every message you see is actually a table with 3 cols:
	//  the time,
	//  the person who caused the event,
	//  and the content
	var messageElement = $(document.createElement("table"));
	messageElement.addClass("message");
	// sanitize
	text = util.toStaticHTML(text);
	var content = '<tr>' + '  <td class="date">' + util.timeString(time) + '</td>' + '  <td class="nick">' + util.toStaticHTML(from) + ' says to ' + name + ': ' + '</td>' + '  <td class="msg-text">' + text + '</td>' + '</tr>';
	messageElement.html(content);
	//the log is the stream that we view
	$("#chatHistory").append(messageElement);
	base += increase;
	scrollDown(base);
}
1190835
function postMessage(channelState, messageJson) {
  var writePromise = writeMessage(channelState.channelName, channelState.uuid, messageJson, channelState.paths);
  channelState.writeBlockPromise = channelState.writeBlockPromise.then(
  /*#__PURE__*/
  _asyncToGenerator(
  /*#__PURE__*/
  _regeneratorRuntime.mark(function _callee3() {
    var _ref4, msgObj, pingStr, writeToReadersPromise;

    return _regeneratorRuntime.wrap(function _callee3$(_context3) {
      while (1) {
        switch (_context3.prev = _context3.next) {
          case 0:
            _context3.next = 2;
            return new Promise(function (res) {
              return setTimeout(res, 0);
            });

          case 2:
            _context3.next = 4;
            return Promise.all([writePromise, refreshReaderClients(channelState)]);

          case 4:
            _ref4 = _context3.sent;
            msgObj = _ref4[0];
            emitOverFastPath(channelState, msgObj, messageJson);
            pingStr = '{"t":' + msgObj.time + ',"u":"' + msgObj.uuid + '","to":"' + msgObj.token + '"}|';
            writeToReadersPromise = Promise.all(Object.values(channelState.otherReaderClients).filter(function (client) {
              return client.writable;
            }) // client might have closed in between
            .map(function (client) {
              return new Promise(function (res) {
                client.write(pingStr, res);
              });
            }));
            /**
             * clean up old messages
             * to not waste resources on cleaning up,
             * only if random-int matches, we clean up old messages
             */

            if (randomInt(0, 20) === 0) {
              /* await */
              getAllMessages(channelState.channelName, channelState.paths).then(function (allMessages) {
                return cleanOldMessages(allMessages, channelState.options.node.ttl);
              });
            }

            return _context3.abrupt("return", writeToReadersPromise);

          case 11:
          case "end":
            return _context3.stop();
        }
      }
    }, _callee3);
  })));
  return channelState.writeBlockPromise;
}
265004
static int fillBufferWithDelimitedMessageFrom(InputStream in,
            boolean drainRemainingBytesIfTooLarge,
            LinkedBuffer lb) throws IOException
    {
        final byte[] buf = lb.buffer;
        int offset = lb.start, len = buf.length - offset, read = in.read(buf, offset, len);

        if (read < 1)
            throw new EOFException("fillBufferWithDelimitedMessageFrom");

        int last = offset + read, size = buf[offset++];
        if (0 != (size & 0x80))
        {
            size = size & 0x7f;

            for (int shift = 7;; shift += 7)
            {
                if (offset == last)
                {
                    // read too few bytes
                    read = in.read(buf, last, len - (last - lb.start));
                    if (read < 1)
                        throw new EOFException("fillBufferWithDelimitedMessageFrom");

                    last += read;
                }

                byte b = buf[offset++];
                size |= (b & 0x7f) << shift;

                if (0 == (b & 0x80))
                    break;

                if (shift == 28)
                {
                    // discard the remaining bytes (5)
                    for (int i = 0;;)
                    {
                        if (offset == last)
                        {
                            // read more
                            read = in.read(buf, last, len - (last - lb.start));
                            if (read < 1)
                                throw new EOFException("fillBufferWithDelimitedMessageFrom");

                            last += read;
                        }

                        if (buf[offset++] >= 0)
                            break;

                        if (5 == ++i)
                        {
                            // we've already consumed 10 bytes
                            throw ProtobufException.malformedVarint();
                        }
                    }

                }
            }
        }

        if (size == 0)
        {
            if (offset != last)
                throw ProtobufException.misreportedSize();

            return size;
        }

        if (size < 0)
            throw ProtobufException.negativeSize();

        final int partial = last - offset;
        if (partial < size)
        {
            // need to read more
            final int delimSize = offset - lb.start;
            if ((size + delimSize) > len)
            {
                // too large.

                if (!drainRemainingBytesIfTooLarge)
                    return size;

                // drain the remaining bytes
                for (int remaining = size - partial; remaining > 0;)
                {
                    read = in.read(buf, lb.start, Math.min(remaining, len));
                    if (read < 1)
                        throw new EOFException("fillBufferWithDelimitedMessageFrom");

                    remaining -= read;
                }

                return size;
            }

            // read the remaining bytes
            fillBufferFrom(in, buf, last, size - partial);
        }

        // set the read offset (start of message)
        lb.offset = offset;

        return size;
    }
438882
function emitData() {
      for(var i=0;i<streamQueue.length;i++) {
        var dataQueue = streamQueue[i].dataQueue;

        if(streamQueue[i].pending) {
          return;
        }

        for(var j=0;j<dataQueue.length;j++) {
          var data = dataQueue[j];
          if(!!data) {
            _this.emit('data', data);
            dataQueue[j] = null;

            return emitData();
          }
        }
      }

      if(currentStream === streamQueue.length) {
        _this.emit('end');
      }
    }
618410
public final void flush() {
        int totalBytes = 0;
        int amountToPoll = outgoingPackets.size();
        
        Packet packet;

        boolean shouldEncrypt = encryption != null;
        
        var queue = new ArrayDeque<byte[]>();

        while (amountToPoll-- > 0 && (packet = outgoingPackets.poll()) != null) {
            int currentBytes = totalBytes;

            boolean tooBig = (totalBytes += packet.getSize(this)) >= bufferSize;
            boolean empty = outgoingPackets.isEmpty();

            if (!tooBig || empty) {
                queue.addAll(packet.getQueue());
            }

            // If we've buffered all of the packets that we can, send them off.
            if (tooBig || empty) {
                var raw = DIRECT_BUFFER_POOL.take(empty ? totalBytes : currentBytes);
                
                byte[] input;
                
                try {
                    while ((input = queue.pollFirst()) != null) {
                        raw.put(shouldEncrypt ? encryption.doFinal(input) : input);
                    }
                } catch (Exception e) {
                    throw new IllegalStateException("An exception occurred whilst encrypting data:", e);
                }
                
                raw.flip();
                
                queue.addAll(packet.getQueue());
                
                // It is important to synchronize on the packetsToFlush here, as we don't want the callback to
                // complete before the packet is queued.
                synchronized (packetsToFlush) {
                    if (!writing.getAndSet(true)) {
                        channel.write(raw, new Tuple<>(this, raw), PACKET_HANDLER);
                    } else {
                        packetsToFlush.offerFirst(raw);
                    }
                }

                totalBytes = packet.getSize(this);
            }
        }
    }
553723
function LineStream(writeStream, readStream) {
    var self = this;
    this.writeStream = writeStream;
    this.readStream = readStream;

    // resume streams
    this.readStream.resume();

    // save data in this buffer
    this.buffer = "";

    // new data from stream will be handled by this
    this.readStream.on('data', function (chunk) {
      self.buffer += chunk;

      // go through the buffer and isolate each message by a linebreak
      var i, start = 0;
      while ((i = self.buffer.indexOf('\n', start)) >= 0) {

        // parse line as JSON and emit line
        self.emit('message', JSON.parse(self.buffer.slice(start, i)));
        start = i + 1;
      }
      self.buffer = self.buffer.slice(start);
    });
  }
536831
function getCumulativeMessages(timestamp, clientSync) {
    var msgs = [];
    if (lastSync > clientSync) {
        msgs = messages;
    } else {
        var newMsgs = messages.length;
        while (newMsgs > 0 && messages[newMsgs - 1].timestamp > timestamp) {
            newMsgs--;
        }
        for (var i = newMsgs; i < messages.length; i++) {
            msgs.push(messages[i]);
        }
    }

    var data = {};
    data.msgs = msgs;
    data.lastSync = lastSync;

    return data;
}
808122
function _handleMessagePing() {
  _handleMessagePing = _asyncToGenerator(
  /*#__PURE__*/
  _regeneratorRuntime.mark(function _callee18(state, msgObj) {
    var messages, useMessages;
    return _regeneratorRuntime.wrap(function _callee18$(_context18) {
      while (1) {
        switch (_context18.prev = _context18.next) {
          case 0:
            if (state.messagesCallback) {
              _context18.next = 2;
              break;
            }

            return _context18.abrupt("return");

          case 2:
            if (msgObj) {
              _context18.next = 8;
              break;
            }

            _context18.next = 5;
            return getAllMessages(state.channelName, state.paths);

          case 5:
            messages = _context18.sent;
            _context18.next = 9;
            break;

          case 8:
            // get single message
            messages = [getSingleMessage(state.channelName, msgObj, state.paths)];

          case 9:
            useMessages = messages.filter(function (msgObj) {
              return _filterMessage(msgObj, state);
            }).sort(function (msgObjA, msgObjB) {
              return msgObjA.time - msgObjB.time;
            }); // sort by time
            // if no listener or message, so not do anything

            if (!(!useMessages.length || !state.messagesCallback)) {
              _context18.next = 12;
              break;
            }

            return _context18.abrupt("return");

          case 12:
            _context18.next = 14;
            return Promise.all(useMessages.map(function (msgObj) {
              return readMessage(msgObj).then(function (content) {
                return msgObj.content = content;
              });
            }));

          case 14:
            useMessages.forEach(function (msgObj) {
              state.emittedMessagesIds.add(msgObj.token);

              if (state.messagesCallback) {
                // emit to subscribers
                state.messagesCallback(msgObj.content.data);
              }
            });

          case 15:
          case "end":
            return _context18.stop();
        }
      }
    }, _callee18);
  }));
  return _handleMessagePing.apply(this, arguments);
}
1324840
function streamToString(callback) {
  var contents = '';

  return through(
    function (chunk, enc, next) {
      contents += chunk.toString('utf8');
      next();
    },
    function (next) { // flush function
      callback.bind(this)(contents, next);
    }
  );
}