write a TaskMessage into a stream

Each TaskMessage is encoded as: task ... short(2) len ... int(4) payload ... byte[] *
344306
private void writeTaskMessage(ChannelBufferOutputStream bout, TaskMessage message) throws Exception {
        int payload_len = 0;
        if (message.message() != null)
            payload_len = message.message().length;

        short type = message.get_type();
        bout.writeShort(type);

        int task_id = message.task();
        if (task_id > Short.MAX_VALUE)
            throw new RuntimeException("Task ID should not exceed " + Short.MAX_VALUE);

        bout.writeShort((short) task_id);
        bout.writeInt(payload_len);
        if (payload_len > 0)
            bout.write(message.message());

        // LOG.info("Write one message taskid:{}, len:{}, data:{}", taskId
        // , payload_len, JStormUtils.toPrintableString(message.message()) );
    }
506031
public void deliver(final byte[] message) {
    synchronized (this.heartBeatManager) {
      if (this.currentStatus.isNotRunning()) {
        LOG.log(Level.WARNING,
            "Trying to send a message to a task that is in state: {0}. Ignoring.",
            this.currentStatus.getState());
      } else {
        try {
          this.deliverMessageToTask(message);
        } catch (final TaskMessageHandlerFailure taskMessageHandlerFailure) {
          LOG.log(Level.WARNING, "Exception while executing task close handler.",
              taskMessageHandlerFailure.getCause());
          this.currentStatus.setException(taskMessageHandlerFailure.getCause());
        }
      }
    }
  }
901622
private void writeArrayMessage(ByteBufAllocator allocator, ArrayRedisMessage msg, List<Object> out) {
        if (msg.isNull()) {
            writeArrayHeader(allocator, msg.isNull(), RedisConstants.NULL_VALUE, out);
        } else {
            writeArrayHeader(allocator, msg.isNull(), msg.children().size(), out);
            for (RedisMessage child : msg.children()) {
                writeRedisMessage(allocator, child, out);
            }
        }
    }
611395
function write(s) {
		var err
		var args = Array.prototype.slice.call(arguments)
		var cb = typeof args[args.length - 1] === 'function' ? args.pop() : null
		var cbCounter = 1
		if (writeToStdoutEnabled) {
			cbCounter++
			process.stdout.write(s, stdWr)
		}
		if (writeStream) {
			cbCounter++
			writeStream.write(s, wRes)
		}
		end()

		function stdWr(e) {
			if (e) {
				if (!err) err = e
				writeToStdoutEnabled = false
				write('stdout error: ' + e.message)
				self.emit('error', e)
			}
			end()
		}

		function wRes(e) {
			if (e) {
				if (!err) err = e
				writeStream = false
				write('stream error: ' + e.message)
				self.emit('error', e)
			}
			end()
		}

		function end() {
			if (!--cbCounter)
				if (cb)
					if (err) cb(err)
					else cb()
		}
	}
949552
public static void writeShort(byte[] bytes, short value, int offset) {
        bytes[offset] = (byte) (0xFF & (value >> 8));
        bytes[offset + 1] = (byte) (0xFF & value);
    }
252017
public void writeBytes(byte[] buffer, int offset, int length)
        throws IOException
    {
        if (buffer == null) {
            if (SIZE < _offset + 16)
                flushBuffer();

            _buffer[_offset++] = (byte) 'N';
        }
        else {
            while (SIZE - _offset - 3 < length) {
                int sublen = SIZE - _offset - 3;

                if (sublen < 16) {
                    flushBuffer();

                    sublen = SIZE - _offset - 3;

                    if (length < sublen)
                        sublen = length;
                }

                _buffer[_offset++] = (byte) BC_BINARY_CHUNK;
                _buffer[_offset++] = (byte) (sublen >> 8);
                _buffer[_offset++] = (byte) sublen;

                System.arraycopy(buffer, offset, _buffer, _offset, sublen);
                _offset += sublen;

                length -= sublen;
                offset += sublen;

                flushBuffer();
            }

            if (SIZE < _offset + 16)
                flushBuffer();

            if (length <= BINARY_DIRECT_MAX) {
                _buffer[_offset++] = (byte) (BC_BINARY_DIRECT + length);
            }
            else if (length <= BINARY_SHORT_MAX) {
                _buffer[_offset++] = (byte) (BC_BINARY_SHORT + (length >> 8));
                _buffer[_offset++] = (byte) (length);
            }
            else {
                _buffer[_offset++] = (byte) 'B';
                _buffer[_offset++] = (byte) (length >> 8);
                _buffer[_offset++] = (byte) (length);
            }

            System.arraycopy(buffer, offset, _buffer, _offset, length);

            _offset += length;
        }
    }
206813
def send_messages(self, messages):
        """"""
        if not messages:
            return
        self._lock.acquire()
        try:
            # The try-except is nested to allow for
            # Python 2.4 support (Refs #12147)
            try:
                stream_created = self.open()
                for message in messages:
                    self.stream.write(render_message(message))
                    self.stream.write('\n')
                    self.stream.write('-'*79)
                    self.stream.write('\n')
                    self.stream.flush()  # flush after each message
                if stream_created:
                    self.close()
            except:
                if not self.fail_silently:
                    raise
        finally:
            self._lock.release()
        return len(messages)
722022
def encode(msg):
    
    msg = bytes(msg)
    return END + msg.replace(ESC, ESC + ESC_ESC).replace(END, ESC + ESC_END) + END
175220
public List<String> encodeTask(final Task<Diff> task)
		throws ConfigurationException, UnsupportedEncodingException,
		DecodingException, EncodingException, SQLConsumerException
	{

		// this.task = task;
		if (task.getTaskType() == TaskTypes.TASK_FULL
				|| task.getTaskType() == TaskTypes.TASK_PARTIAL_FIRST) {

			this.lastFullRevID = -1;
		}

		int articleId = task.getHeader().getArticleId();
		Diff diff;

		ArrayList<String> list = new ArrayList<String>();

		String tempData = new String();

		int size = task.size();
		for (int i = 0; i < size; i++) {

			diff = task.get(i);

			if (diff.isFullRevision()) {
				this.lastFullRevID = diff.getRevisionID();
			}

			/*
			 * prepare values that might be null
			 * because we don't want quotes if they are null
			 *
			 * Furthermore, escape quote-characters. Quotes are used as the "ENCLOSED BY" character
			 * in MySQL to mark begin and end of Strings
			 */

			//prepare values that might be null
			//because we don't want quotes if they are null
			String comm = diff.getComment();
			String comment = comm==null?"\\N":"\""+escape(comm)+"\"";

			Integer cId = diff.getContributorId();
			String contributorId = cId==null?"\\N":cId.toString();

			String cName = diff.getContributorName();
			String contributorName = cName==null?"\\N":"\""+escape(cName)+"\"";

			//Prepare the actual data item
			tempData = "\\N,"
					+ this.lastFullRevID + ","
					+ diff.getRevisionCounter() + ","
					+ diff.getRevisionID()+ ","
					+ articleId + ","
					+ diff.getTimeStamp().getTime()+ ",\""
					+ encodeDiff(task, diff) + "\","
					+ comment+","
					+ (diff.isMinor()?"1":"0")+","
					+ contributorName+","
					+ contributorId+ ","
					+ (diff.getContributorIsRegistered()?"1":"0");

			//add item to the list
			list.add(tempData);
		}

		return list;
	}
1211303
@Override
  public byte[] toBytes(TaskKey taskKey) {
    return Bytes.add(jobKeyConv.toBytes(taskKey), Constants.SEP_BYTES,
        Bytes.toBytes(taskKey.getTaskId()));
  }
1140958
async def _write_to_container_stdin(self, write_stream, message):
        
        msg = msgpack.dumps(message, encoding="utf8", use_bin_type=True)
        self._logger.debug("Sending %i bytes to container", len(msg))
        write_stream.write(struct.pack('I', len(msg)))
        write_stream.write(msg)
        await write_stream.drain()
425999
public void
writeByteArray(byte [] b, int off, int len) {
	need(len);
	System.arraycopy(b, off, array, pos, len);
	pos += len;
}
758958
private void sendMessage(BeanMessageID type, Buffer payload) {
        Buffer buffer = new Buffer();
        buffer.writeByte((type.getRawValue() >> 8) & 0xff);
        buffer.writeByte(type.getRawValue() & 0xff);
        if (payload != null) {
            try {
                buffer.writeAll(payload);
            } catch (IOException e) {
                throw new RuntimeException(e);
            }
        }
        GattSerialMessage serialMessage = GattSerialMessage.fromPayload(buffer.readByteArray());
        gattClient.getSerialProfile().sendMessage(serialMessage.getBuffer());
    }
613607
def do_write(self, msg):
        """"""
        self.stream = self.do_open()
        stream = self.stream
        stream.write(msg)
        if self.terminator:
            stream.write(self.terminator)
        stream.flush()
        self._close()
        return
145915
protected void sendInBound(int taskId, HeronTuples.HeronTupleSet message) {
    if (message.hasData()) {
      sendMessageToInstance(taskId, message);
    }

    if (message.hasControl()) {
      processAcksAndFails(message.getSrcTaskId(), taskId, message.getControl());
    }
  }
939895
private void writeObject(ObjectOutputStream out) throws IOException {
        final boolean trace = TraceComponent.isAnyTracingEnabled();
        if (trace && tc.isEntryEnabled())

            Tr.entry(this, tc, "writeObject", new Object[] {
                                                            nonserTaskClassName,
                                                            nonserTriggerClassName,
                                                            "initial delay " + initialDelay,
                                                            "interval " + interval,
                                                            isFixedRate ? "fixed rate" : false,
                                                            submittedAsCallable ? "callable" : "runnable",
                                                            threadContext
            });

        if (threadContextBytes == null)
            threadContextBytes = threadContext == null ? null : threadContext.serialize();

        PutField fields = out.putFields();
        fields.put(INITIAL_DELAY, initialDelay);
        fields.put(INTERVAL, interval);
        fields.put(IS_FIXED_RATE, isFixedRate);
        fields.put(NONSER_TASK_CLASS_NAME, nonserTaskClassName);
        fields.put(NONSER_TRIGGER_CLASS_NAME, nonserTriggerClassName);
        fields.put(SUBMITTED_AS_CALLABLE, submittedAsCallable);
        fields.put(THREAD_CONTEXT_DESCRIPTOR, threadContextBytes);
        out.writeFields();

        if (trace && tc.isEntryEnabled())
            Tr.exit(this, tc, "writeObject");
    }
235091
@Override
  public void streamReply(WebSocket session,
                          HeadersAmp headers,
                          String to,
                          long qid,
                          int sequence,
                          List<Object> values,
                          Throwable exn,
                          boolean isComplete)
    throws IOException
  {
    /*
    if (values != null && values.size() > 0) {
      try (Writer out = session.getBasicRemote().getSendWriter()) {
        init(out);
        streamResult(headers, to, qid, values);
      }
    }
    
    if (exn != null) {
      try (Writer out = session.getBasicRemote().getSendWriter()) {
        init(out);
        streamError(headers, to, qid, exn);
      }
    }
    else if (isComplete) {
      try (Writer out = session.getBasicRemote().getSendWriter()) {
        init(out);
        streamComplete(headers, to, qid);
      }
    }
    */
  }