rpc_request_task::rpc_request_task(message_ptr& request, service_node* node) 
    : task(task_code(request->header().local_rpc_code), request->header().client.hash, node), 
    _request(request)
{

    dbg_dassert (TASK_TYPE_RPC_REQUEST == spec().type, "task type must be RPC_REQUEST, please use DEFINE_TASK_CODE_RPC to define the task code");
}

int replication_app_base::dispatch_rpc_call(int code, message_ptr& request, bool ack_client)
{
    auto it = _handlers.find(code);
    if (it != _handlers.end())
    {
        if (ack_client)
        {
            message_ptr response = request->create_response();
            int err = 0;
            marshall(response->writer(), err);
            it->second(request, response);
        }
        else
        {
            message_ptr response(nullptr);
            it->second(request, response);
        }
    }
    else if (ack_client)
    {
        message_ptr response = request->create_response();
        error_code err = ERR_HANDLER_NOT_FOUND;
        marshall(response->writer(), (int)err);
        rpc::reply(response);
    }

    return 0;
}

void on_message(client* c, websocketpp::connection_hdl hdl, message_ptr msg) {
    client::connection_ptr con = sip_client.get_con_from_hdl(hdl);

    std::cout << "Received a reply:" << std::endl;
    fwrite(msg->get_payload().c_str(), msg->get_payload().size(), 1, stdout);
    received=true;
}

// Define a callback to handle incoming messages
void on_message(server* s, websocketpp::connection_hdl hdl, message_ptr msg) {
    try {
        s->send(hdl, msg->get_payload(), msg->get_opcode());
    } catch (const websocketpp::lib::error_code& e) {

    }
}

 void on_message(connection_ptr con, message_ptr msg) {        
     if (con->get_resource() == "/getcasecount") {
         std::cout << "detected " << msg->get_payload() << " test cases." << std::endl;
         m_case_count = atoi(msg->get_payload().c_str());
     } else {
         con->send(msg->get_payload(),msg->get_opcode());
     }
 }

    void rpc_server_session::on_recv_request(message_ptr& msg, int delay_ms)
    {
        msg->header().from_address = remote_address();
        msg->header().from_address.port = msg->header().client.port;
        msg->header().to_address = _net.address();

        msg->server_session().reset(this);
        return _net.on_recv_request(msg, delay_ms);
    }

void websocketIO::on_message(websocketpp::connection_hdl hdl, message_ptr msg) {
    boost::property_tree::ptree json;
    std::istringstream iss(msg->get_payload());
    boost::property_tree::read_json(iss, json);
    
    std::string func = json.get<std::string> ("func");
    if(messages.find(func) != messages.end()){
        messages[func](this, msg->get_payload());
    }
}

// Define a callback to handle incoming messages
void on_message(server* s, websocketpp::connection_hdl hdl, message_ptr msg) {
    std::cout << " and message: " << msg->get_payload() <<std::endl ;
    std::cout << " opcode " << msg->get_opcode() <<std::endl ;

    //    s->send(hdl, msg->get_payload(), msg->get_opcode());

    const char* msge = "{\"GESTURE\":\"Move Left\"}";

    s->send(hdl, msge, websocketpp::frame::opcode::text);
}

   void Handler::on_message(connection_ptr con, message_ptr msg)
   {
      if (msg->get_opcode() != websocketpp::frame::opcode::TEXT)
         return;

      Log::debug("client <%s>: <message> %s", plain(con).c_str(),
            msg->get_payload().c_str());

      publish(format_message(msg->get_payload()));
   }

    void sim_server_session::send(message_ptr& reply_msg)
    {
        message_ptr recv_msg(new message(reply_msg->writer().get_buffer()));
        recv_msg->header().from_address = reply_msg->header().from_address;
        recv_msg->header().to_address = reply_msg->header().to_address;

        _client->on_recv_reply(recv_msg->header().id, recv_msg,
            recv_msg->header().from_address == recv_msg->header().to_address ?
            0 : (static_cast<sim_network_provider*>(&_net))->net_delay_milliseconds()
            );
    }

void on_message(client* c, websocketpp::connection_hdl hdl, message_ptr msg) {
    client::connection_ptr con = c->get_con_from_hdl(hdl);

    if (con->get_resource() == "/getCaseCount") {
        std::cout << "Detected " << msg->get_payload() << " test cases."
                  << std::endl;
        case_count = atoi(msg->get_payload().c_str());
    } else {
        c->send(hdl, msg->get_payload(), msg->get_opcode());
    }
}

void on_message(server* s, websocketpp::connection_hdl hdl, message_ptr msg) {
    std::cout << "on_message called with hdl: " << hdl.lock().get()
              << " and message: " << msg->get_payload()
              << std::endl;

    try {
        s->send(hdl, msg->get_payload(), msg->get_opcode());
    } catch (const websocketpp::lib::error_code& e) {
        std::cout << "Echo failed because: " << e
                  << "(" << e.message() << ")" << std::endl;
    }
}

void ReadUncompressed::handleRx(const message_ptr message)
{
    if (!m_initialized)
    {
        m_initialized = check();
        if (!m_initialized) m_error = true;
    }
    else
    {
        if (message->get_opcode() == websocketpp::frame::opcode::binary)
        {
            const std::string& bytes(message->get_payload());
            const std::size_t rawNumBytes(bytes.size());
            const std::size_t stride(m_layout->pointSize());

            m_data.insert(m_data.end(), bytes.begin(), bytes.end());

            const std::size_t wholePoints(m_data.size() / stride);

            PointId nextId(m_view->size());
            const PointId doneId(nextId + wholePoints);

            const char* pos(m_data.data());

            while (nextId < doneId)
            {
                for (const auto& dim : m_layout->dims())
                {
                    m_view->setField(
                            dim,
                            m_layout->dimType(dim),
                            nextId,
                            pos);

                    pos += m_layout->dimSize(dim);
                }

                ++nextId;
            }

            m_numBytesReceived += rawNumBytes;
            m_data.assign(
                    m_data.begin() + wholePoints * stride,
                    m_data.end());
        }
        else
        {
            m_error = true;
        }
    }
}

void websocket_server::on_message(ws_socket* socket_, websocketpp::connection_hdl hdl, message_ptr msg) {
    
    if(msg->get_payload() == "CC"){
        cc_hdl_ = hdl;
        clientConnected = true;
        BOOST_LOG_TRIVIAL(info) << "Control Center connected via websocket";
    }
    else if (msg->get_payload() == "AL"){
        al_hdl_ = hdl;
        proxyConnected = true;
        BOOST_LOG_TRIVIAL(info) << "Al-Proxy connected via websocket";
    }
    
}

void replica::response_client_message(message_ptr& request, error_code error, decree d/* = invalid_decree*/)
{
    if (nullptr == request)
        return;

    message_ptr resp = request->create_response();
    resp->writer().write(error);

    dassert(error != ERR_OK, "");
    dinfo("handle replication request with rpc_id = %016llx failed, err = %s",
        request->header().rpc_id, error.to_string());

    rpc::reply(resp);
}

void chat_server_handler::on_message(connection_ptr con, message_ptr msg) {
    if (msg->get_opcode() != websocketpp::frame::opcode::text) {
        return;
    }
    
    std::cout << "message from client " << con << ": " << msg->get_payload() << std::endl;
    
    
    
    // check for special command messages
    if (msg->get_payload() == "/help") {
        // print command list
        con->send(encode_message("server","avaliable commands:<br />&nbsp;&nbsp;&nbsp;&nbsp;/help - show this help<br />&nbsp;&nbsp;&nbsp;&nbsp;/alias foo - set alias to foo",false));
        return;
    }
    
    if (msg->get_payload().substr(0,7) == "/alias ") {
        std::string response;
        std::string alias;
        
        if (msg->get_payload().size() == 7) {
            response = "you must enter an alias.";
            con->send(encode_message("server",response));
            return;
        } else {
            alias = msg->get_payload().substr(7);
        }
        
        response = m_connections[con] + " is now known as "+alias;

        // store alias pre-escaped so we don't have to do this replacing every time this
        // user sends a message
        
        // escape json characters
        boost::algorithm::replace_all(alias,"\\","\\\\");
        boost::algorithm::replace_all(alias,"\"","\\\"");
        
        // escape html characters
        boost::algorithm::replace_all(alias,"&","&amp;");
        boost::algorithm::replace_all(alias,"<","&lt;");
        boost::algorithm::replace_all(alias,">","&gt;");
        
        m_connections[con] = alias;
        
        // set alias
        send_to_all(serialize_state());
        send_to_all(encode_message("server",response));
        return;
    }
    
    // catch other slash commands
    if ((msg->get_payload())[0] == '/') {
        con->send(encode_message("server","unrecognized command"));
        return;
    }
    
    // create json message to send based on msg
    send_to_all(encode_message(m_connections[con],msg->get_payload()));
}

    lib::error_code prepare_close(close::status::value code,
        std::string const & reason, message_ptr out) const
    {
        if (!out) {
            return lib::error_code(error::invalid_arguments);
        }

        std::string val;
        val.append(1,'\xff');
        val.append(1,'\x00');
        out->set_payload(val);
        out->set_prepared(true);

        return lib::error_code();
    }

void mutation::set_client_request(task_code code, message_ptr& request)
{
    dassert(client_request == nullptr, "batch is not supported now");
    client_request = request;
    rpc_code = code;
    data.updates.push_back(request->reader().get_remaining_buffer());
}

    void command_manager::on_remote_cli(message_ptr& request)
    {
        std::string cmd;
        unmarshall(request->reader(), cmd);

        std::vector<std::string> args;
        unmarshall(request->reader(), args);

        std::string result;
        run_command(cmd, args, result);

        auto resp = request->create_response();
        marshall(resp->writer(), result);

        ::dsn::service::rpc::reply(resp);
    }

void WSConnection::rcvMessage (message_ptr msg, bool& msgRejected, bool& runQueue)
{
    ScopedLockType sl (m_receiveQueueMutex);

    if (m_isDead)
    {
        msgRejected = false;
        runQueue = false;
        return;
    }

    if ((m_receiveQueue.size () >= 1000) || (msg->get_payload().size() > 1000000))
    {
        msgRejected = true;
        runQueue = false;
    }
    else
    {
        msgRejected = false;
        m_receiveQueue.push_back (msg);

        if (m_receiveQueueRunning)
            runQueue = false;
        else
        {
            runQueue = true;
            m_receiveQueueRunning = true;
        }
    }
}