def METHOD_NAME(self, task: TxTask, final=False):
    if task.buffer_size == 0:
        payload = None
    elif task.buffer_size == STREAM_CHUNK_SIZE:
        if task.direct_buf:
            payload = task.direct_buf
        else:
            payload = task.buffer
    else:
        payload = wrap_view(task.buffer)[0 : task.buffer_size]
    message = Message(None, payload)
    if task.offset == 0:
        # User headers are only included in the first chunk
        if task.headers:
            message.add_headers(task.headers)
        message.add_headers(
            {
                StreamHeaderKey.CHANNEL: task.channel,
                StreamHeaderKey.TOPIC: task.topic,
                StreamHeaderKey.SIZE: task.stream.get_size(),
            }
        )
    message.add_headers(
        {
            StreamHeaderKey.STREAM_ID: task.sid,
            StreamHeaderKey.DATA_TYPE: StreamDataType.FINAL if final else StreamDataType.CHUNK,
            StreamHeaderKey.SEQUENCE: task.seq,
            StreamHeaderKey.OFFSET: task.offset,
        }
    )
    errors = self.cell.fire_and_forget(STREAM_CHANNEL, STREAM_DATA_TOPIC, task.target, message)
    error = errors.get(task.target)
    if error:
        msg = f"Message sending error to target {task.target}: {error}"
        log.debug(msg)
        self._stop_task(task, StreamError(msg))
        return
    # Update state
    task.seq += 1
    task.offset += task.buffer_size
    task.buffer_size = 0
    task.direct_buf = None
    # Update future
    task.stream_future.set_progress(task.offset)