Method CutMessageIntoFileDescriptor

src/brpc/stream.cpp:148–215

}

ssize_t Stream::CutMessageIntoFileDescriptor(int /*fd*/, 
                                             butil::IOBuf **data_list, 
                                             size_t size) {
    if (_host_socket == NULL) {
        CHECK(false) << "Not connected";
        errno = EBADF;
        return -1;
    }
    if (!_remote_settings.writable()) {
        LOG(WARNING) << "The remote side of Stream=" << id() 
                     << "->" << _remote_settings.stream_id()
                     << "@" << _host_socket->remote_side()
                     << " doesn't have a handler";
        errno = EBADF;
        return -1;
    }
    butil::IOBuf out;
    ssize_t len = 0;
    ssize_t unwritten_data_size = 0;
    for (size_t i = 0; i < size; ++i) {
        butil::IOBuf *data = data_list[i];
        size_t length = data->length();
        if (length > FLAGS_stream_write_max_segment_size) {
            if (unwritten_data_size) {
                WriteToHostSocket(&out);
                unwritten_data_size = 0;
                out.clear();
            }
            // segmenting large data into multiple parts
            butil::IOBuf segment_buf;
            bool has_continuation = true;
            while (has_continuation) {
                data->cutn(&segment_buf, FLAGS_stream_write_max_segment_size);
                StreamFrameMeta fm;
                fm.set_stream_id(_remote_settings.stream_id());
                fm.set_source_stream_id(id());
                fm.set_frame_type(FRAME_TYPE_DATA);
                has_continuation = !data->empty();
                fm.set_has_continuation(has_continuation);
                policy::PackStreamMessage(&out, fm, &segment_buf);
                len += segment_buf.length();
                segment_buf.clear();
                WriteToHostSocket(&out);
                out.clear();
            }
        } else {
            if (unwritten_data_size + length > FLAGS_stream_write_max_segment_size) {
                WriteToHostSocket(&out);
                unwritten_data_size = 0;
                out.clear();
            }
            unwritten_data_size += length;
            StreamFrameMeta fm;
            fm.set_stream_id(_remote_settings.stream_id());
            fm.set_source_stream_id(id());
            fm.set_frame_type(FRAME_TYPE_DATA);
            fm.set_has_continuation(false);
            policy::PackStreamMessage(&out, fm, data_list[i]);