void tcp_adapter_proxy::async_setup_web_socket_write_buffer_drain()

in src/TcpAdapterProxy.cpp [1456:1508]

• 44 lines of code
• 5 McCabe index (conditional complexity)

    void tcp_adapter_proxy::async_setup_web_socket_write_buffer_drain(tcp_adapter_context &tac, std::string const & service_id)
    {
        BOOST_LOG_SEV(log, trace) << "Web socket write buffer drain for service id: " << service_id;
        boost::beast::flat_buffer                   outgoing_message_buffer;
        tcp_connection::pointer connection = get_tcp_connection(tac, service_id);
        using namespace com::amazonaws::iot::securedtunneling;
        if (connection->web_socket_data_write_buffer_.size() > 0)
        {
            // Get end point from the tcp socket

            outgoing_message.set_type(Message_Type_DATA);
            if (tac.adapter_config.serviceId_to_endpoint_map.find(service_id) == tac.adapter_config.serviceId_to_endpoint_map.end())
            {
                throw proxy_exception((boost::format("Could not forward traffic from invalid service id: %1%") % service_id).str());
            }
            else if (tac.serviceId_to_streamId_map.find(service_id) == tac.serviceId_to_streamId_map.end())
            {
                throw proxy_exception((boost::format("No streamId exists for the service Id %1%") % service_id).str());
            }
            BOOST_LOG_SEV(log, debug) << "Prepare to send data message: service id: " << service_id << " stream id: " << tac.serviceId_to_streamId_map[service_id];
            // Construct outgoing message
            outgoing_message.set_serviceid(service_id);
            outgoing_message.set_streamid(tac.serviceId_to_streamId_map[service_id]);
            size_t const send_size = std::min<std::size_t>(GET_SETTING(settings, MESSAGE_MAX_PAYLOAD_SIZE),
                                                           connection->web_socket_data_write_buffer_.size());
            boost::asio::buffer_copy(outgoing_message_buffer.prepare(send_size), connection->web_socket_data_write_buffer_.data(), send_size);
            outgoing_message_buffer.commit(send_size);
            outgoing_message.set_payload(outgoing_message_buffer.data().data(), send_size);

            // Clean up web_socket_data_write_buffer
            connection->web_socket_data_write_buffer_.consume(send_size);
            outgoing_message_buffer.consume(outgoing_message_buffer.max_size());

            //after message is sent, continue with the loop
            connection->after_send_message = std::bind(&tcp_adapter_proxy::async_setup_web_socket_write_buffer_drain, this, std::ref(tac), service_id);
            async_send_message(tac, outgoing_message);

            //if this write cleared up enough space
            if (wss_has_enough_write_buffer_space(connection->web_socket_data_write_buffer_))
            {
                BOOST_LOG_SEV(log, debug) << "Write buffer has enough space, continue tcp read loop for " << service_id ;
                async_tcp_socket_read_loop(tac, service_id);
            }
            else
            {
                BOOST_LOG_SEV(log, debug) << " write DOES NOT cleared up enough space, no tcp read loop" << service_id ;
            }
        }
        else
        {   //not writing, no buffer contents, skip straight to being done draining
            invoke_and_clear_handler(connection->on_web_socket_write_buffer_drain_complete);
        }
    }