// Licensed to the Apache Software Foundation (ASF) under one
// or more contributor license agreements.  See the NOTICE file
// distributed with this work for additional information
// regarding copyright ownership.  The ASF licenses this file
// to you under the Apache License, Version 2.0 (the
// "License"); you may not use this file except in compliance
// with the License.  You may obtain a copy of the License at
//
//   http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing,
// software distributed under the License is distributed on an
// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
// KIND, either express or implied.  See the License for the
// specific language governing permissions and limitations
// under the License.

// brpc - A framework to host and access services throughout Baidu.

// Date: Sun Jul 13 15:04:18 CST 2014

#include <sys/types.h>
#include <sys/socket.h>
#include <gtest/gtest.h>
#include <gflags/gflags.h>
#include <google/protobuf/descriptor.h>
#include "butil/time.h"
#include "butil/macros.h"
#include "butil/logging.h"
#include "butil/files/temp_file.h"
#include "brpc/socket.h"
#include "brpc/acceptor.h"
#include "brpc/server.h"
#include "brpc/policy/baidu_rpc_protocol.h"
#include "brpc/policy/baidu_rpc_meta.pb.h"
#include "brpc/policy/most_common_message.h"
#include "brpc/channel.h"
#include "brpc/details/load_balancer_with_naming.h"
#include "brpc/parallel_channel.h"
#include "brpc/selective_channel.h"
#include "brpc/socket_map.h"
#include "brpc/controller.h"
#include "echo.pb.h"
#include "brpc/options.pb.h"

namespace brpc {
DECLARE_int32(idle_timeout_second);
DECLARE_int32(max_connection_pool_size);
class Server;
class Span;
class MethodStatus;
namespace policy {
void SendRpcResponse(int64_t correlation_id,
                     Controller* cntl,
                     RpcPBMessages* messages,
                     const Server* server_raw,
                     MethodStatus *, int64_t,
                     std::shared_ptr<Span> span);
} // policy
} // brpc

int main(int argc, char* argv[]) {
    brpc::FLAGS_idle_timeout_second = 0;
    brpc::FLAGS_max_connection_pool_size = 0;
    testing::InitGoogleTest(&argc, argv);
    GFLAGS_NAMESPACE::ParseCommandLineFlags(&argc, &argv, true);
    return RUN_ALL_TESTS();
}

namespace {
void* RunClosure(void* arg) {
    google::protobuf::Closure* done = (google::protobuf::Closure*)arg;
    done->Run();
    return NULL;
}

void MarkCalled(bool* called) {
    *called = true;
}

class DeleteOnlyOnceChannel : public brpc::Channel {
public:
    DeleteOnlyOnceChannel() : _c(1) {
    }
    ~DeleteOnlyOnceChannel() {
        RELEASE_ASSERT_VERBOSE(_c.fetch_sub(1) == 1,
                               "Delete more than once!");
    }
private:
    butil::atomic<int> _c;
};

static std::string MOCK_CREDENTIAL = "mock credential";
static std::string MOCK_CONTEXT = "mock context";

class MyAuthenticator : public brpc::Authenticator {
public:
    MyAuthenticator() : count(0) {}

    int GenerateCredential(std::string* auth_str) const {
        *auth_str = MOCK_CREDENTIAL;
        count.fetch_add(1, butil::memory_order_relaxed);
        return 0;
    }

    int VerifyCredential(const std::string&,
                         const butil::EndPoint&,
                         brpc::AuthContext* ctx) const {
        ctx->set_user(MOCK_CONTEXT);
        ctx->set_group(MOCK_CONTEXT);
        ctx->set_roles(MOCK_CONTEXT);
        ctx->set_starter(MOCK_CONTEXT);
        ctx->set_is_service(true);
        return 0;
    }
    mutable butil::atomic<int32_t> count;
};

static bool VerifyMyRequest(const brpc::InputMessageBase* msg_base) {
    const brpc::policy::MostCommonMessage* msg = 
        static_cast<const brpc::policy::MostCommonMessage*>(msg_base);
    brpc::Socket* ptr = msg->socket();
    
    brpc::policy::RpcMeta meta;
    butil::IOBufAsZeroCopyInputStream wrapper(msg->meta);
    EXPECT_TRUE(meta.ParseFromZeroCopyStream(&wrapper));

    if (meta.has_authentication_data()) {
        // Credential MUST only appear in the first packet
        EXPECT_TRUE(NULL == ptr->auth_context());
        EXPECT_EQ(meta.authentication_data(), MOCK_CREDENTIAL);
        MyAuthenticator authenticator;
        return authenticator.VerifyCredential(
            "", butil::EndPoint(), ptr->mutable_auth_context()) == 0;
    }
    return true;
}

class CallAfterRpcObject {
public:
    explicit CallAfterRpcObject() {}

    ~CallAfterRpcObject() {
        EXPECT_EQ(str, "CallAfterRpcRespTest");
    }

    void Append(const std::string& s) {
        str.append(s);
    }

private:
    std::string str;
};

class MyEchoService : public ::test::EchoService {
    void Echo(google::protobuf::RpcController* cntl_base,
              const ::test::EchoRequest* req,
              ::test::EchoResponse* res,
              google::protobuf::Closure* done) {
        brpc::Controller* cntl =
            static_cast<brpc::Controller*>(cntl_base);
        std::shared_ptr<CallAfterRpcObject> str_test(new CallAfterRpcObject());
        cntl->set_after_rpc_resp_fn(std::bind(&MyEchoService::CallAfterRpc, str_test,
            std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));
        brpc::ClosureGuard done_guard(done);
        if (req->server_fail()) {
            cntl->SetFailed(req->server_fail(), "Server fail1");
            cntl->SetFailed(req->server_fail(), "Server fail2");
            return;
        }
        if (req->close_fd()) {
            LOG(INFO) << "close fd...";
            cntl->CloseConnection("Close connection according to request");
            return;
        }
        if (req->sleep_us() > 0) {
            LOG(INFO) << "sleep " << req->sleep_us() << "us...";
            bthread_usleep(req->sleep_us());
        }
        res->set_message("received " + req->message());
        if (req->code() != 0) {
            res->add_code_list(req->code());
        }
        res->set_receiving_socket_id(cntl->_current_call.sending_sock->id());
        if (mockfunc_) mockfunc_(cntl_base, req, res, done);

        brpc::ProtocolType protocol = cntl->request_protocol();
        if ((brpc::PROTOCOL_HTTP == protocol || brpc::PROTOCOL_H2 == protocol) &&
            !req->http_header().empty()) {
            ASSERT_FALSE(req->http_header().empty());
            const std::string* val = cntl->http_request().GetHeader(req->http_header());
            ASSERT_TRUE(val);
            ASSERT_FALSE(val->empty());
            cntl->http_response().SetHeader(req->http_header(), *val);
        }
    }
    static void CallAfterRpc(std::shared_ptr<CallAfterRpcObject> str,
                        brpc::Controller* cntl,
                        const google::protobuf::Message* req,
                        const google::protobuf::Message* res) {
        const test::EchoRequest* request = static_cast<const test::EchoRequest*>(req);
        const test::EchoResponse* response = static_cast<const test::EchoResponse*>(res);
        str->Append("CallAfterRpcRespTest");
        EXPECT_TRUE(nullptr != cntl);
        EXPECT_TRUE(nullptr != request);
        EXPECT_TRUE(nullptr != response);
    }

public:
    using MockFuncType = void(google::protobuf::RpcController*,
                    const ::test::EchoRequest*, ::test::EchoResponse*,
                    google::protobuf::Closure*);
    void SetMockFunc(std::function<MockFuncType>&& mockfunc) {
        mockfunc_ = std::move(mockfunc);
    }

private:
    std::function<MockFuncType> mockfunc_;
};

class DelayedCloseEchoService : public ::test::EchoService {
public:
    void Echo(google::protobuf::RpcController* cntl_base,
              const ::test::EchoRequest* req,
              ::test::EchoResponse*,
              google::protobuf::Closure* done) override {
        brpc::ClosureGuard done_guard(done);
        if (req->sleep_us() > 0) {
            bthread_usleep(req->sleep_us());
        }
        static_cast<brpc::Controller*>(cntl_base)->CloseConnection(
            "Close connection after delay");
    }
};

pthread_once_t register_mock_protocol = PTHREAD_ONCE_INIT;

class ChannelTest : public ::testing::Test{
protected:
    ChannelTest() 
        : _ep(butil::IP_ANY, 8787)
        , _close_fd_once(false) {
        if (!_dummy.options().rpc_pb_message_factory) {
            _dummy._options.rpc_pb_message_factory = new brpc::DefaultRpcPBMessageFactory();
        }

        pthread_once(&register_mock_protocol, register_protocol);
        const brpc::InputMessageHandler pairs[] = {
            { brpc::policy::ParseRpcMessage, 
              ProcessRpcRequest, VerifyMyRequest, this, "baidu_std" }
        };
        EXPECT_EQ(0, _messenger.AddHandler(pairs[0]));

        EXPECT_EQ(0, _server_list.save(butil::endpoint2str(_ep).c_str()));           
        _naming_url = std::string("File://") + _server_list.fname();
    };

    virtual ~ChannelTest(){};
    virtual void SetUp() {
    };
    virtual void TearDown() {
        StopAndJoin();
    };

    static void register_protocol() {
        brpc::Protocol dummy_protocol = 
                                 { brpc::policy::ParseRpcMessage,
                                   brpc::SerializeRequestDefault, 
                                   brpc::policy::PackRpcRequest,
                                   NULL, ProcessRpcRequest,
                                   VerifyMyRequest, NULL, NULL,
                                   brpc::CONNECTION_TYPE_ALL, "baidu_std" };
        ASSERT_EQ(0,  RegisterProtocol((brpc::ProtocolType)30, dummy_protocol));
    }

    static void ProcessRpcRequest(brpc::InputMessageBase* msg_base) {
        brpc::DestroyingPtr<brpc::policy::MostCommonMessage> msg(
            static_cast<brpc::policy::MostCommonMessage*>(msg_base));
        brpc::SocketUniquePtr ptr(msg->ReleaseSocket());
        const brpc::AuthContext* auth = ptr->auth_context();
        if (auth) {
            EXPECT_EQ(MOCK_CONTEXT, auth->user());
            EXPECT_EQ(MOCK_CONTEXT, auth->group());
            EXPECT_EQ(MOCK_CONTEXT, auth->roles());
            EXPECT_EQ(MOCK_CONTEXT, auth->starter());
            EXPECT_TRUE(auth->is_service());
        }
        ChannelTest* ts = (ChannelTest*)msg_base->arg();
        if (ts->_close_fd_once) {
            ts->_close_fd_once = false;
            ptr->SetFailed();
            return;
        }
        
        brpc::policy::RpcMeta meta;
        butil::IOBufAsZeroCopyInputStream wrapper(msg->meta);
        EXPECT_TRUE(meta.ParseFromZeroCopyStream(&wrapper));
        const brpc::policy::RpcRequestMeta& req_meta = meta.request();
        ASSERT_EQ(ts->_svc.descriptor()->full_name(), req_meta.service_name());
        const google::protobuf::MethodDescriptor* method =
            ts->_svc.descriptor()->FindMethodByName(req_meta.method_name());
        brpc::RpcPBMessages* messages =
            ts->_dummy.options().rpc_pb_message_factory->Get(ts->_svc, *method);
        google::protobuf::Message* req = messages->Request();
        google::protobuf::Message* res = messages->Response();
        if (meta.attachment_size() != 0) {
            butil::IOBuf req_buf;
            msg->payload.cutn(&req_buf, msg->payload.size() - meta.attachment_size());
            butil::IOBufAsZeroCopyInputStream wrapper2(req_buf);
            EXPECT_TRUE(req->ParseFromZeroCopyStream(&wrapper2));
        } else {
            butil::IOBufAsZeroCopyInputStream wrapper2(msg->payload);
            EXPECT_TRUE(req->ParseFromZeroCopyStream(&wrapper2));
        }
        brpc::Controller* cntl = new brpc::Controller();
        cntl->_current_call.peer_id = ptr->id();
        cntl->_current_call.sending_sock.reset(ptr.release());
        cntl->_server = &ts->_dummy;

        google::protobuf::Closure* done = brpc::NewCallback<
            int64_t, brpc::Controller*, brpc::RpcPBMessages*, const brpc::Server*,
            brpc::MethodStatus*, int64_t, std::shared_ptr<brpc::Span>>(
                &brpc::policy::SendRpcResponse, meta.correlation_id(), cntl,
                messages, &ts->_dummy, NULL, -1, nullptr);
        ts->_svc.CallMethod(method, cntl, req, res, done);
    }

    int StartAccept(butil::EndPoint ep) {
        int listening_fd = -1;
        while ((listening_fd = tcp_listen(ep)) < 0) {
            if (errno == EADDRINUSE) {
                bthread_usleep(1000);
            } else {
                return -1;
            }
        }
        if (_messenger.StartAccept(listening_fd, -1, NULL, false) != 0) {
            return -1;
        }
        return 0;
    }

    void StopAndJoin() {
        _messenger.StopAccept(0);
        _messenger.Join();
    }

    void SetUpChannel(brpc::Channel* channel, 
                      bool single_server,
                      bool short_connection,
                      const brpc::Authenticator* auth = NULL,
                      std::string connection_group = std::string(),
                      bool use_backup_request_policy = false,
                      brpc::ProtocolType protocol = brpc::PROTOCOL_BAIDU_STD) {
        brpc::ChannelOptions opt;
        opt.protocol = protocol;
        if (short_connection) {
            opt.connection_type = brpc::CONNECTION_TYPE_SHORT;
        }
        opt.auth = auth;
        opt.max_retry = 0;
        opt.connection_group = connection_group;
        if (use_backup_request_policy) {
            opt.backup_request_policy = &_backup_request_policy;
        }
        if (single_server) {
            EXPECT_EQ(0, channel->Init(_ep, &opt)); 
        } else {                                                 
            EXPECT_EQ(0, channel->Init(_naming_url.c_str(), "rR", &opt));
        }                                         
    }
    
    void CallMethod(brpc::ChannelBase* channel, 
                    brpc::Controller* cntl,
                    test::EchoRequest* req, test::EchoResponse* res,
                    bool async, bool destroy = false) {
        google::protobuf::Closure* done = NULL;                     
        brpc::CallId sync_id = { 0 };
        if (async) {
            sync_id = cntl->call_id();
            done = brpc::DoNothing();
        }
        ::test::EchoService::Stub(channel).Echo(cntl, req, res, done);
        if (async) {
            if (destroy) {
                delete channel;
            }
            // Callback MUST be called for once and only once
            bthread_id_join(sync_id);
        }
    }

    void CallMethod(brpc::ChannelBase* channel, 
                    brpc::Controller* cntl,
                    test::ComboRequest* req, test::ComboResponse* res,
                    bool async, bool destroy = false) {
        google::protobuf::Closure* done = NULL;
        brpc::CallId sync_id = { 0 };
        if (async) {
            sync_id = cntl->call_id();
            done = brpc::DoNothing();
        }
        ::test::EchoService::Stub(channel).ComboEcho(cntl, req, res, done);
        if (async) {
            if (destroy) {
                delete channel;
            }
            // Callback MUST be called for once and only once
            bthread_id_join(sync_id);
        }
    }

    void TestConnectionFailed(bool single_server, bool async, 
                              bool short_connection) {
        std::cout << " *** single=" << single_server
                  << " async=" << async
                  << " short=" << short_connection << std::endl;

        brpc::Channel channel;
        SetUpChannel(&channel, single_server, short_connection);

        brpc::Controller cntl;
        test::EchoRequest req;
        test::EchoResponse res;
        req.set_message(__FUNCTION__);
        CallMethod(&channel, &cntl, &req, &res, async);
        
        EXPECT_EQ(ECONNREFUSED, cntl.ErrorCode()) << cntl.ErrorText();
    }
    
    void TestConnectionFailedParallel(bool single_server, bool async, 
                                      bool short_connection) {
        std::cout << " *** single=" << single_server
                  << " async=" << async
                  << " short=" << short_connection << std::endl;

        const size_t NCHANS = 8;
        brpc::Channel subchans[NCHANS];
        brpc::ParallelChannel channel;
        for (size_t i = 0; i < NCHANS; ++i) {
            SetUpChannel(&subchans[i], single_server, short_connection);
            ASSERT_EQ(0, channel.AddChannel(
                          &subchans[i], brpc::DOESNT_OWN_CHANNEL,
                          NULL, NULL));
        }

        brpc::Controller cntl;
        test::EchoRequest req;
        test::EchoResponse res;
        req.set_message(__FUNCTION__);
        CallMethod(&channel, &cntl, &req, &res, async);
        
        EXPECT_TRUE(brpc::ETOOMANYFAILS == cntl.ErrorCode() ||
                    ECONNREFUSED == cntl.ErrorCode()) << cntl.ErrorText();
        LOG(INFO) << cntl.ErrorText();
    }

    void TestConnectionFailedSelective(bool single_server, bool async, 
                                       bool short_connection) {
        std::cout << " *** single=" << single_server
                  << " async=" << async
                  << " short=" << short_connection << std::endl;

        const size_t NCHANS = 8;
        brpc::SelectiveChannel channel;
        brpc::ChannelOptions options;
        options.max_retry = 0;
        ASSERT_EQ(0, channel.Init("rr", &options));
        for (size_t i = 0; i < NCHANS; ++i) {
            brpc::Channel* subchan = new brpc::Channel;
            SetUpChannel(subchan, single_server, short_connection);
            ASSERT_EQ(0, channel.AddChannel(subchan, NULL)) << "i=" << i;
        }

        brpc::Controller cntl;
        test::EchoRequest req;
        test::EchoResponse res;
        req.set_message(__FUNCTION__);
        CallMethod(&channel, &cntl, &req, &res, async);
        
        EXPECT_EQ(ECONNREFUSED, cntl.ErrorCode()) << cntl.ErrorText();
        ASSERT_EQ(1, cntl.sub_count());
        EXPECT_EQ(ECONNREFUSED, cntl.sub(0)->ErrorCode())
            << cntl.sub(0)->ErrorText();
        LOG(INFO) << cntl.ErrorText();
    }
    
    void TestSuccess(bool single_server, bool async, bool short_connection) {
        std::cout << " *** single=" << single_server
                  << " async=" << async
                  << " short=" << short_connection << std::endl;

        ASSERT_EQ(0, StartAccept(_ep));
        brpc::Channel channel;
        SetUpChannel(&channel, single_server, short_connection);
                
        brpc::Controller cntl;
        test::EchoRequest req;
        test::EchoResponse res;
        req.set_message(__FUNCTION__);
        CallMethod(&channel, &cntl, &req, &res, async);

        EXPECT_EQ(0, cntl.ErrorCode()) 
            << single_server << ", " << async << ", " << short_connection;
        const uint64_t receiving_socket_id = res.receiving_socket_id();
        EXPECT_EQ(0, cntl.sub_count());
        EXPECT_TRUE(NULL == cntl.sub(-1));
        EXPECT_TRUE(NULL == cntl.sub(0));
        EXPECT_TRUE(NULL == cntl.sub(1));
        EXPECT_EQ("received " + std::string(__FUNCTION__), res.message());
        if (short_connection) {
            // Sleep to let `_messenger' detect `Socket' being `SetFailed'
            const int64_t start_time = butil::cpuwide_time_us();
            while (_messenger.ConnectionCount() != 0) {
                EXPECT_LT(butil::cpuwide_time_us(), start_time + 100000L/*100ms*/);
                bthread_usleep(1000);
            }
        } else {
            EXPECT_GE(1ul, _messenger.ConnectionCount());
        }
        if (single_server && !short_connection) {
            // Reuse the connection
            brpc::Channel channel2;
            SetUpChannel(&channel2, single_server, short_connection);
            cntl.Reset();
            req.Clear();
            res.Clear();
            req.set_message(__FUNCTION__);
            CallMethod(&channel2, &cntl, &req, &res, async);
            EXPECT_EQ(0, cntl.ErrorCode())
                << single_server << ", " << async << ", " << short_connection;
            EXPECT_EQ(receiving_socket_id, res.receiving_socket_id());

            // A different connection_group does not reuse the connection
            brpc::Channel channel3;
            SetUpChannel(&channel3, single_server, short_connection,
                         NULL, "another_group");
            cntl.Reset();
            req.Clear();
            res.Clear();
            req.set_message(__FUNCTION__);
            CallMethod(&channel3, &cntl, &req, &res, async);
            EXPECT_EQ(0, cntl.ErrorCode())
                << single_server << ", " << async << ", " << short_connection;
            const uint64_t receiving_socket_id2 = res.receiving_socket_id();
            EXPECT_NE(receiving_socket_id, receiving_socket_id2);

            // Channel in the same connection_group reuses the connection
            // note that the leading/trailing spaces should be trimed.
            brpc::Channel channel4;
            SetUpChannel(&channel4, single_server, short_connection,
                         NULL, " another_group ");
            cntl.Reset();
            req.Clear();
            res.Clear();
            req.set_message(__FUNCTION__);
            CallMethod(&channel4, &cntl, &req, &res, async);
            EXPECT_EQ(0, cntl.ErrorCode())
                << single_server << ", " << async << ", " << short_connection;
            EXPECT_EQ(receiving_socket_id2, res.receiving_socket_id());
        }
        StopAndJoin();
    }

    class SetCode : public brpc::CallMapper {
    public:
        brpc::SubCall Map(
            int channel_index,
            const google::protobuf::MethodDescriptor* method,
            const google::protobuf::Message* req_base,
            google::protobuf::Message* response) override {
            test::EchoRequest* req = brpc::Clone<test::EchoRequest>(req_base);
            req->set_code(channel_index + 1/*non-zero*/);
            return brpc::SubCall(method, req, response->New(),
                                brpc::DELETE_REQUEST | brpc::DELETE_RESPONSE);
        }
    };

    class SetCodeOnEven : public SetCode {
    public:
        brpc::SubCall Map(
            int channel_index,
            const google::protobuf::MethodDescriptor* method,
            const google::protobuf::Message* req_base,
            google::protobuf::Message* response) override {
            if (channel_index % 2) {
                return brpc::SubCall::Skip();
            }
            return SetCode::Map(channel_index, method, req_base, response);
        }
    };


    class GetReqAndAddRes : public brpc::CallMapper {
        brpc::SubCall Map(
            int channel_index,
            const google::protobuf::MethodDescriptor* method,
            const google::protobuf::Message* req_base,
            google::protobuf::Message* res_base) override {
            const test::ComboRequest* req =
                dynamic_cast<const test::ComboRequest*>(req_base);
            test::ComboResponse* res = dynamic_cast<test::ComboResponse*>(res_base);
            if (method->name() != "ComboEcho" ||
                res == NULL || req == NULL ||
                req->requests_size() <= channel_index) {
                return brpc::SubCall::Bad();
            }
            return brpc::SubCall(::test::EchoService::descriptor()->method(0),
                                &req->requests(channel_index),
                                res->add_responses(), 0);
        }
    };

    class SuccessLimitCallMapper : public brpc::CallMapper {
    public:
        brpc::SubCall Map(int channel_index,
                          const google::protobuf::MethodDescriptor* method,
                          const google::protobuf::Message* req_base,
                          google::protobuf::Message* response) override {
            auto req = brpc::Clone<test::EchoRequest>(req_base);
            req->set_code(channel_index + 1/*non-zero*/);
            if (_index++ > 0) {
                req->set_sleep_us(5 * 1000);
            }
            return brpc::SubCall(method, req, response->New(),
                                 brpc::DELETE_REQUEST | brpc::DELETE_RESPONSE);
        }
    private:
        size_t _index{0};
    };

    class MergeNothing : public brpc::ResponseMerger {
        Result Merge(google::protobuf::Message* /*response*/,
                     const google::protobuf::Message* /*sub_response*/) {
            return brpc::ResponseMerger::MERGED;
        }
    };

    void TestSuccessParallel(bool single_server, bool async, bool short_connection) {
        std::cout << " *** single=" << single_server
                  << " async=" << async
                  << " short=" << short_connection << std::endl;

        ASSERT_EQ(0, StartAccept(_ep));
        const size_t NCHANS = 8;
        brpc::Channel subchans[NCHANS];
        brpc::ParallelChannel channel;
        for (size_t i = 0; i < NCHANS; ++i) {
            SetUpChannel(&subchans[i], single_server, short_connection);
            ASSERT_EQ(0, channel.AddChannel(
                          &subchans[i], brpc::DOESNT_OWN_CHANNEL,
                          new SetCode, NULL));
        }
        brpc::Controller cntl;
        test::EchoRequest req;
        test::EchoResponse res;
        req.set_message(__FUNCTION__);
        req.set_code(23);
        CallMethod(&channel, &cntl, &req, &res, async);

        EXPECT_EQ(0, cntl.ErrorCode()) << cntl.ErrorText();
        EXPECT_EQ(NCHANS, (size_t)cntl.sub_count());
        for (int i = 0; i < cntl.sub_count(); ++i) {
            EXPECT_TRUE(cntl.sub(i) && !cntl.sub(i)->Failed()) << "i=" << i;
        }
        EXPECT_EQ("received " + std::string(__FUNCTION__), res.message());
        ASSERT_EQ(NCHANS, (size_t)res.code_list_size());
        for (size_t i = 0; i < NCHANS; ++i) {
            ASSERT_EQ((int)i+1, res.code_list(i));
        }
        if (short_connection) {
            // Sleep to let `_messenger' detect `Socket' being `SetFailed'
            const int64_t start_time = butil::cpuwide_time_us();
            while (_messenger.ConnectionCount() != 0) {
                EXPECT_LT(butil::cpuwide_time_us(), start_time + 100000L/*100ms*/);
                bthread_usleep(1000);
            }
        } else {
            EXPECT_GE(1ul, _messenger.ConnectionCount());
        }
        StopAndJoin();
    }

    void TestSuccessDuplicatedParallel(
        bool single_server, bool async, bool short_connection) {
        std::cout << " *** single=" << single_server
                  << " async=" << async
                  << " short=" << short_connection << std::endl;

        ASSERT_EQ(0, StartAccept(_ep));
        const size_t NCHANS = 8;
        brpc::Channel* subchan = new DeleteOnlyOnceChannel;
        SetUpChannel(subchan, single_server, short_connection);
        brpc::ParallelChannel channel;
        // Share the CallMapper and ResponseMerger should be fine because
        // they're intrusively shared.
        SetCode* set_code = new SetCode;
        for (size_t i = 0; i < NCHANS; ++i) {
            ASSERT_EQ(0, channel.AddChannel(
                          subchan,
                          // subchan should be deleted (for only once)
                          ((i % 2) ? brpc::DOESNT_OWN_CHANNEL : brpc::OWNS_CHANNEL),
                          set_code, NULL));
        }
        ASSERT_EQ((int)NCHANS, set_code->ref_count());
        brpc::Controller cntl;
        test::EchoRequest req;
        test::EchoResponse res;
        req.set_message(__FUNCTION__);
        req.set_code(23);
        CallMethod(&channel, &cntl, &req, &res, async);

        EXPECT_EQ(0, cntl.ErrorCode()) << cntl.ErrorText();
        EXPECT_EQ(NCHANS, (size_t)cntl.sub_count());
        for (int i = 0; i < cntl.sub_count(); ++i) {
            EXPECT_TRUE(cntl.sub(i) && !cntl.sub(i)->Failed()) << "i=" << i;
        }
        EXPECT_EQ("received " + std::string(__FUNCTION__), res.message());
        ASSERT_EQ(NCHANS, (size_t)res.code_list_size());
        for (size_t i = 0; i < NCHANS; ++i) {
            ASSERT_EQ((int)i+1, res.code_list(i));
        }
        if (short_connection) {
            // Sleep to let `_messenger' detect `Socket' being `SetFailed'
            const int64_t start_time = butil::cpuwide_time_us();
            while (_messenger.ConnectionCount() != 0) {
                EXPECT_LT(butil::cpuwide_time_us(), start_time + 100000L/*100ms*/);
                bthread_usleep(1000);
            }
        } else {
            EXPECT_GE(1ul, _messenger.ConnectionCount());
        }
        StopAndJoin();
    }
    
    void TestSuccessSelective(bool single_server, bool async, bool short_connection) {
        std::cout << " *** single=" << single_server
                  << " async=" << async
                  << " short=" << short_connection << std::endl;

        const size_t NCHANS = 8;
        ASSERT_EQ(0, StartAccept(_ep));
        brpc::SelectiveChannel channel;
        brpc::ChannelOptions options;
        options.max_retry = 0;
        ASSERT_EQ(0, channel.Init("rr", &options));
        for (size_t i = 0; i < NCHANS; ++i) {
            brpc::Channel* subchan = new brpc::Channel;
            SetUpChannel(subchan, single_server, short_connection);
            ASSERT_EQ(0, channel.AddChannel(subchan, NULL)) << "i=" << i;
        }
        brpc::Controller cntl;
        test::EchoRequest req;
        test::EchoResponse res;
        req.set_message(__FUNCTION__);
        req.set_code(23);
        CallMethod(&channel, &cntl, &req, &res, async);

        EXPECT_EQ(0, cntl.ErrorCode()) << cntl.ErrorText();
        EXPECT_EQ(1, cntl.sub_count());
        ASSERT_EQ(0, cntl.sub(0)->ErrorCode());
        EXPECT_EQ("received " + std::string(__FUNCTION__), res.message());
        ASSERT_EQ(1, res.code_list_size());
        ASSERT_EQ(req.code(), res.code_list(0));
        ASSERT_EQ(_ep, cntl.remote_side());
        
        if (short_connection) {
            // Sleep to let `_messenger' detect `Socket' being `SetFailed'
            const int64_t start_time = butil::cpuwide_time_us();
            while (_messenger.ConnectionCount() != 0) {
                EXPECT_LT(butil::cpuwide_time_us(), start_time + 100000L/*100ms*/);
                bthread_usleep(1000);
            }
        } else {
            EXPECT_GE(1ul, _messenger.ConnectionCount());
        }
        StopAndJoin();
    }

    void TestSkipParallel(bool single_server, bool async, bool short_connection) {
        std::cout << " *** single=" << single_server
                  << " async=" << async
                  << " short=" << short_connection << std::endl;

        ASSERT_EQ(0, StartAccept(_ep));
        const size_t NCHANS = 8;
        brpc::Channel subchans[NCHANS];
        brpc::ParallelChannel channel;
        for (size_t i = 0; i < NCHANS; ++i) {
            SetUpChannel(&subchans[i], single_server, short_connection);
            ASSERT_EQ(0, channel.AddChannel(
                          &subchans[i], brpc::DOESNT_OWN_CHANNEL,
                          new SetCodeOnEven, NULL));
        }
        brpc::Controller cntl;
        test::EchoRequest req;
        test::EchoResponse res;
        req.set_message(__FUNCTION__);
        req.set_code(23);
        CallMethod(&channel, &cntl, &req, &res, async);

        EXPECT_EQ(0, cntl.ErrorCode()) << cntl.ErrorText();
        EXPECT_EQ("received " + std::string(__FUNCTION__), res.message());
        EXPECT_EQ(NCHANS, (size_t)cntl.sub_count());
        for (int i = 0; i < cntl.sub_count(); ++i) {
            if (i % 2) {
                EXPECT_TRUE(NULL == cntl.sub(i)) << "i=" << i;
            } else {
                EXPECT_TRUE(cntl.sub(i) && !cntl.sub(i)->Failed()) << "i=" << i;
            }
        }
        ASSERT_EQ(NCHANS / 2, (size_t)res.code_list_size());
        for (int i = 0; i < res.code_list_size(); ++i) {
            ASSERT_EQ(i*2 + 1, res.code_list(i));
        }
        if (short_connection) {
            // Sleep to let `_messenger' detect `Socket' being `SetFailed'
            const int64_t start_time = butil::cpuwide_time_us();
            while (_messenger.ConnectionCount() != 0) {
                EXPECT_LT(butil::cpuwide_time_us(), start_time + 100000L/*100ms*/);
                bthread_usleep(1000);
            }
        } else {
            EXPECT_GE(1ul, _messenger.ConnectionCount());
        }
        StopAndJoin();
    }

    void TestSuccessParallel2(bool single_server, bool async, bool short_connection) {
        std::cout << " *** single=" << single_server
                  << " async=" << async
                  << " short=" << short_connection << std::endl;

        ASSERT_EQ(0, StartAccept(_ep));
        const size_t NCHANS = 8;
        brpc::Channel subchans[NCHANS];
        brpc::ParallelChannel channel;
        for (size_t i = 0; i < NCHANS; ++i) {
            SetUpChannel(&subchans[i], single_server, short_connection);
            ASSERT_EQ(0, channel.AddChannel(
                          &subchans[i], brpc::DOESNT_OWN_CHANNEL,
                          new GetReqAndAddRes, new MergeNothing));
        }
        brpc::Controller cntl;
        test::ComboRequest req;
        test::ComboResponse res;
        CallMethod(&channel, &cntl, &req, &res, false);
        ASSERT_TRUE(cntl.Failed()); // req does not have .requests
        ASSERT_EQ(brpc::EREQUEST, cntl.ErrorCode());

        for (size_t i = 0; i < NCHANS; ++i) {
            ::test::EchoRequest* sub_req = req.add_requests();
            sub_req->set_message(butil::string_printf("hello_%llu", (long long)i));
            sub_req->set_code(i + 1);
        }

        // non-parallel channel does not work.
        cntl.Reset();
        CallMethod(&subchans[0], &cntl, &req, &res, false);
        ASSERT_TRUE(cntl.Failed());
        ASSERT_EQ(brpc::EINTERNAL, cntl.ErrorCode()) << cntl.ErrorText();
        ASSERT_TRUE(butil::StringPiece(cntl.ErrorText()).ends_with("Method ComboEcho() not implemented."));

        // do the rpc call.
        cntl.Reset();
        CallMethod(&channel, &cntl, &req, &res, async);

        EXPECT_EQ(0, cntl.ErrorCode()) << cntl.ErrorText();
        ASSERT_GT(cntl.latency_us(), 0);
        ASSERT_EQ((int)NCHANS, res.responses_size());
        for (int i = 0; i < res.responses_size(); ++i) {
            EXPECT_EQ(butil::string_printf("received hello_%d", i),
                      res.responses(i).message());
            ASSERT_EQ(1, res.responses(i).code_list_size());
            EXPECT_EQ(i + 1, res.responses(i).code_list(0));
        }
        if (short_connection) {
            // Sleep to let `_messenger' detect `Socket' being `SetFailed'
            const int64_t start_time = butil::cpuwide_time_us();
            while (_messenger.ConnectionCount() != 0) {
                EXPECT_LT(butil::cpuwide_time_us(), start_time + 100000L/*100ms*/);
                bthread_usleep(1000);
            }
        } else {
            EXPECT_GE(1ul, _messenger.ConnectionCount());
        }
        StopAndJoin();
    }

    void TestSuccessLimitParallel(bool single_server, bool async, bool short_connection) {
        std::cout << " *** single=" << single_server
                  << " async=" << async
                  << " short=" << short_connection << std::endl;

        ASSERT_EQ(0, StartAccept(_ep));
        const size_t NCHANS = 8;
        brpc::Channel subchans[NCHANS];
        brpc::ParallelChannel channel;
        brpc::ParallelChannelOptions options;
        // Only care about the first successful response.
        options.success_limit = 1;
        channel.Init(&options);
        butil::intrusive_ptr<brpc::CallMapper> fast_call_mapper(new SuccessLimitCallMapper);
        for (size_t i = 0; i < NCHANS; ++i) {
            SetUpChannel(&subchans[i], single_server, short_connection);
            ASSERT_EQ(0, channel.AddChannel(
                &subchans[i], brpc::DOESNT_OWN_CHANNEL, fast_call_mapper, NULL));
        }
        brpc::Controller cntl;
        test::EchoRequest req;
        test::EchoResponse res;
        req.set_message(__FUNCTION__);
        req.set_code(23);
        CallMethod(&channel, &cntl, &req, &res, async);

        EXPECT_EQ(0, cntl.ErrorCode()) << cntl.ErrorText();
        EXPECT_EQ(NCHANS, (size_t)cntl.sub_count());
        for (int i = 0; i < cntl.sub_count(); ++i) {
            EXPECT_TRUE(cntl.sub(i)) << "i=" << i;
            if (0 == i) {
                EXPECT_TRUE(!cntl.sub(i)->Failed()) << "i=" << i;
            } else {
                EXPECT_TRUE(cntl.sub(i)->Failed()) << "i=" << i;
                EXPECT_EQ(brpc::EPCHANFINISH, cntl.sub(i)->ErrorCode()) << "i=" << i;
            }
        }
        EXPECT_EQ("received " + std::string(__FUNCTION__), res.message());
        ASSERT_EQ(1, res.code_list_size());
        ASSERT_EQ((int)1, res.code_list(0));
        if (short_connection) {
            // Sleep to let `_messenger' detect `Socket' being `SetFailed'
            const int64_t start_time = butil::cpuwide_time_us();
            while (_messenger.ConnectionCount() != 0) {
                EXPECT_LT(butil::cpuwide_time_us(), start_time + 100000L/*100ms*/);
                bthread_usleep(1000);
            }
        } else {
            EXPECT_GE(1ul, _messenger.ConnectionCount());
        }
        StopAndJoin();
    }

    struct CancelerArg {
        int64_t sleep_before_cancel_us;
        brpc::CallId cid;
    };

    static void* Canceler(void* void_arg) {
        CancelerArg* arg = static_cast<CancelerArg*>(void_arg);
        if (arg->sleep_before_cancel_us > 0) {
            bthread_usleep(arg->sleep_before_cancel_us);
        }
        LOG(INFO) << "Start to cancel cid=" << arg->cid.value;
        brpc::StartCancel(arg->cid);
        return NULL;
    }


    void CancelBeforeCallMethod(
        bool single_server, bool async, bool short_connection) {
        std::cout << " *** single=" << single_server
                  << " async=" << async
                  << " short=" << short_connection << std::endl;

        ASSERT_EQ(0, StartAccept(_ep));
        brpc::Channel channel;
        SetUpChannel(&channel, single_server, short_connection);
                
        brpc::Controller cntl;
        test::EchoRequest req;
        test::EchoResponse res;
        req.set_message(__FUNCTION__);
        const brpc::CallId cid = cntl.call_id();
        ASSERT_TRUE(cid.value != 0);
        brpc::StartCancel(cid);
        CallMethod(&channel, &cntl, &req, &res, async);
        EXPECT_EQ(ECANCELED, cntl.ErrorCode()) << cntl.ErrorText();
        StopAndJoin();
    }

    void CancelBeforeCallMethodParallel(
        bool single_server, bool async, bool short_connection) {
        std::cout << " *** single=" << single_server
                  << " async=" << async
                  << " short=" << short_connection << std::endl;

        ASSERT_EQ(0, StartAccept(_ep));

        const size_t NCHANS = 8;
        brpc::Channel subchans[NCHANS];
        brpc::ParallelChannel channel;
        for (size_t i = 0; i < NCHANS; ++i) {
            SetUpChannel(&subchans[i], single_server, short_connection);
            ASSERT_EQ(0, channel.AddChannel(
                          &subchans[i], brpc::DOESNT_OWN_CHANNEL,
                          NULL, NULL));
        }
                
        brpc::Controller cntl;
        test::EchoRequest req;
        test::EchoResponse res;
        req.set_message(__FUNCTION__);
        const brpc::CallId cid = cntl.call_id();
        ASSERT_TRUE(cid.value != 0);
        brpc::StartCancel(cid);
        CallMethod(&channel, &cntl, &req, &res, async);
        EXPECT_EQ(ECANCELED, cntl.ErrorCode()) << cntl.ErrorText();
        EXPECT_EQ(NCHANS, (size_t)cntl.sub_count());
        EXPECT_TRUE(NULL == cntl.sub(1));
        EXPECT_TRUE(NULL == cntl.sub(0));
        StopAndJoin();
    }

    void CancelBeforeCallMethodSelective(
        bool single_server, bool async, bool short_connection) {
        std::cout << " *** single=" << single_server
                  << " async=" << async
                  << " short=" << short_connection << std::endl;

        ASSERT_EQ(0, StartAccept(_ep));

        const size_t NCHANS = 8;
        brpc::SelectiveChannel channel;
        ASSERT_EQ(0, channel.Init("rr", NULL));
        for (size_t i = 0; i < NCHANS; ++i) {
            brpc::Channel* subchan = new brpc::Channel;
            SetUpChannel(subchan, single_server, short_connection);
            ASSERT_EQ(0, channel.AddChannel(subchan, NULL)) << "i=" << i;
        }
                
        brpc::Controller cntl;
        test::EchoRequest req;
        test::EchoResponse res;
        req.set_message(__FUNCTION__);
        const brpc::CallId cid = cntl.call_id();
        ASSERT_TRUE(cid.value != 0);
        brpc::StartCancel(cid);
        CallMethod(&channel, &cntl, &req, &res, async);
        EXPECT_EQ(ECANCELED, cntl.ErrorCode()) << cntl.ErrorText();
        StopAndJoin();
    }

    void CancelDuringCallMethod(
        bool single_server, bool async, bool short_connection) {
        std::cout << " *** single=" << single_server
                  << " async=" << async
                  << " short=" << short_connection << std::endl;

        ASSERT_EQ(0, StartAccept(_ep));
        brpc::Channel channel;
        SetUpChannel(&channel, single_server, short_connection);
                
        brpc::Controller cntl;
        test::EchoRequest req;
        test::EchoResponse res;
        req.set_message(__FUNCTION__);
        const brpc::CallId cid = cntl.call_id();
        ASSERT_TRUE(cid.value != 0);
        pthread_t th;
        CancelerArg carg = { 10000, cid };
        ASSERT_EQ(0, pthread_create(&th, NULL, Canceler, &carg));
        req.set_sleep_us(carg.sleep_before_cancel_us * 2);
        butil::Timer tm;
        tm.start();
        CallMethod(&channel, &cntl, &req, &res, async);
        tm.stop();
        EXPECT_LT(labs(tm.u_elapsed() - carg.sleep_before_cancel_us), 10000);
        ASSERT_EQ(0, pthread_join(th, NULL));
        EXPECT_EQ(ECANCELED, cntl.ErrorCode());
        EXPECT_EQ(0, cntl.sub_count());
        EXPECT_TRUE(NULL == cntl.sub(1));
        EXPECT_TRUE(NULL == cntl.sub(0));
        StopAndJoin();
    }
    
    void CancelDuringCallMethodParallel(
        bool single_server, bool async, bool short_connection) {
        std::cout << " *** single=" << single_server
                  << " async=" << async
                  << " short=" << short_connection << std::endl;

        ASSERT_EQ(0, StartAccept(_ep));

        const size_t NCHANS = 8;
        brpc::Channel subchans[NCHANS];
        brpc::ParallelChannel channel;
        for (size_t i = 0; i < NCHANS; ++i) {
            SetUpChannel(&subchans[i], single_server, short_connection);
            ASSERT_EQ(0, channel.AddChannel(
                          &subchans[i], brpc::DOESNT_OWN_CHANNEL,
                          NULL, NULL));
        }
                
        brpc::Controller cntl;
        test::EchoRequest req;
        test::EchoResponse res;
        req.set_message(__FUNCTION__);
        const brpc::CallId cid = cntl.call_id();
        ASSERT_TRUE(cid.value != 0);
        pthread_t th;
        CancelerArg carg = { 10000, cid };
        ASSERT_EQ(0, pthread_create(&th, NULL, Canceler, &carg));
        req.set_sleep_us(carg.sleep_before_cancel_us * 2);
        butil::Timer tm;
        tm.start();
        CallMethod(&channel, &cntl, &req, &res, async);
        tm.stop();
        EXPECT_LT(labs(tm.u_elapsed() - carg.sleep_before_cancel_us), 10000);
        ASSERT_EQ(0, pthread_join(th, NULL));
        EXPECT_EQ(ECANCELED, cntl.ErrorCode());
        EXPECT_EQ(NCHANS, (size_t)cntl.sub_count());
        for (int i = 0; i < cntl.sub_count(); ++i) {
            EXPECT_EQ(ECANCELED, cntl.sub(i)->ErrorCode()) << "i=" << i;
        }
        EXPECT_LT(labs(cntl.latency_us() - carg.sleep_before_cancel_us), 10000);
        StopAndJoin();
    }

    void CancelDuringCallMethodSelective(
        bool single_server, bool async, bool short_connection) {
        std::cout << " *** single=" << single_server
                  << " async=" << async
                  << " short=" << short_connection << std::endl;

        ASSERT_EQ(0, StartAccept(_ep));

        const size_t NCHANS = 8;
        brpc::SelectiveChannel channel;
        ASSERT_EQ(0, channel.Init("rr", NULL));
        for (size_t i = 0; i < NCHANS; ++i) {
            brpc::Channel* subchan = new brpc::Channel;
            SetUpChannel(subchan, single_server, short_connection);
            ASSERT_EQ(0, channel.AddChannel(subchan, NULL)) << "i=" << i;
        }
                
        brpc::Controller cntl;
        test::EchoRequest req;
        test::EchoResponse res;
        req.set_message(__FUNCTION__);
        const brpc::CallId cid = cntl.call_id();
        ASSERT_TRUE(cid.value != 0);
        pthread_t th;
        CancelerArg carg = { 10000, cid };
        ASSERT_EQ(0, pthread_create(&th, NULL, Canceler, &carg));
        req.set_sleep_us(carg.sleep_before_cancel_us * 2);
        butil::Timer tm;
        tm.start();
        CallMethod(&channel, &cntl, &req, &res, async);
        tm.stop();
        EXPECT_LT(labs(tm.u_elapsed() - carg.sleep_before_cancel_us), 10000);
        ASSERT_EQ(0, pthread_join(th, NULL));
        EXPECT_EQ(ECANCELED, cntl.ErrorCode());
        EXPECT_EQ(1, cntl.sub_count());
        EXPECT_EQ(ECANCELED, cntl.sub(0)->ErrorCode());
        StopAndJoin();
    }
    
    void CancelAfterCallMethod(
        bool single_server, bool async, bool short_connection) {
        std::cout << " *** single=" << single_server
                  << " async=" << async
                  << " short=" << short_connection << std::endl;

        ASSERT_EQ(0, StartAccept(_ep));
        brpc::Channel channel;
        SetUpChannel(&channel, single_server, short_connection);
                
        brpc::Controller cntl;
        test::EchoRequest req;
        test::EchoResponse res;
        req.set_message(__FUNCTION__);
        const brpc::CallId cid = cntl.call_id();
        ASSERT_TRUE(cid.value != 0);
        CallMethod(&channel, &cntl, &req, &res, async);
        EXPECT_EQ(0, cntl.ErrorCode());
        EXPECT_EQ(0, cntl.sub_count());
        ASSERT_EQ(EINVAL, bthread_id_error(cid, ECANCELED));
        StopAndJoin();
    }

    void CancelAfterCallMethodParallel(
        bool single_server, bool async, bool short_connection) {
        std::cout << " *** single=" << single_server
                  << " async=" << async
                  << " short=" << short_connection << std::endl;

        ASSERT_EQ(0, StartAccept(_ep));

        const size_t NCHANS = 8;
        brpc::Channel subchans[NCHANS];
        brpc::ParallelChannel channel;
        for (size_t i = 0; i < NCHANS; ++i) {
            SetUpChannel(&subchans[i], single_server, short_connection);
            ASSERT_EQ(0, channel.AddChannel(
                          &subchans[i], brpc::DOESNT_OWN_CHANNEL,
                          NULL, NULL));
        }
                
        brpc::Controller cntl;
        test::EchoRequest req;
        test::EchoResponse res;
        req.set_message(__FUNCTION__);
        const brpc::CallId cid = cntl.call_id();
        ASSERT_TRUE(cid.value != 0);
        CallMethod(&channel, &cntl, &req, &res, async);
        EXPECT_EQ(0, cntl.ErrorCode());
        EXPECT_EQ(NCHANS, (size_t)cntl.sub_count());
        for (int i = 0; i < cntl.sub_count(); ++i) {
            EXPECT_TRUE(cntl.sub(i) && !cntl.sub(i)->Failed()) << "i=" << i;
        }
        ASSERT_EQ(EINVAL, bthread_id_error(cid, ECANCELED));
        StopAndJoin();
    }

    void TestAttachment(bool async, bool short_connection) {
        ASSERT_EQ(0, StartAccept(_ep));
        brpc::Channel channel;
        SetUpChannel(&channel, true, short_connection);
                
        brpc::Controller cntl;
        cntl.request_attachment().append("attachment");
        test::EchoRequest req;
        test::EchoResponse res;
        req.set_message(__FUNCTION__);
        CallMethod(&channel, &cntl, &req, &res, async);
        
        EXPECT_EQ(0, cntl.ErrorCode())  << short_connection;
        EXPECT_FALSE(cntl.request_attachment().empty())
            << ", " << async << ", " << short_connection;
        EXPECT_EQ("received " + std::string(__FUNCTION__), res.message());
        if (short_connection) {
            // Sleep to let `_messenger' detect `Socket' being `SetFailed'
            const int64_t start_time = butil::cpuwide_time_us();
            while (_messenger.ConnectionCount() != 0) {
                EXPECT_LT(butil::cpuwide_time_us(), start_time + 100000L/*100ms*/);
                bthread_usleep(1000);
            }
        } else {
            EXPECT_GE(1ul, _messenger.ConnectionCount());
        }            
        StopAndJoin();
    }

    void TestRequestNotInit(bool single_server, bool async,
                            bool short_connection) {
        std::cout << " *** single=" << single_server
                  << " async=" << async
                  << " short=" << short_connection << std::endl;
        ASSERT_EQ(0, StartAccept(_ep));
        brpc::Channel channel;
        SetUpChannel(&channel, single_server, short_connection);
                
        brpc::Controller cntl;
        test::EchoRequest req;
        test::EchoResponse res;
        CallMethod(&channel, &cntl, &req, &res, async);
        EXPECT_EQ(brpc::EREQUEST, cntl.ErrorCode()) << cntl.ErrorText();
        StopAndJoin();
    }

    void TestRequestNotInitParallel(bool single_server, bool async,
                                    bool short_connection) {
        std::cout << " *** single=" << single_server
                  << " async=" << async
                  << " short=" << short_connection << std::endl;
        ASSERT_EQ(0, StartAccept(_ep));

        const size_t NCHANS = 8;
        brpc::Channel subchans[NCHANS];
        brpc::ParallelChannel channel;
        for (size_t i = 0; i < NCHANS; ++i) {
            SetUpChannel(&subchans[i], single_server, short_connection);
            ASSERT_EQ(0, channel.AddChannel(
                          &subchans[i], brpc::DOESNT_OWN_CHANNEL,
                          NULL, NULL));
        }
        
        brpc::Controller cntl;
        test::EchoRequest req;
        test::EchoResponse res;
        CallMethod(&channel, &cntl, &req, &res, async);
        EXPECT_EQ(brpc::EREQUEST, cntl.ErrorCode()) << cntl.ErrorText();
        LOG(WARNING) << cntl.ErrorText();
        StopAndJoin();
    }

    void TestRequestNotInitSelective(bool single_server, bool async,
                                     bool short_connection) {
        std::cout << " *** single=" << single_server
                  << " async=" << async
                  << " short=" << short_connection << std::endl;
        ASSERT_EQ(0, StartAccept(_ep));

        const size_t NCHANS = 8;
        brpc::SelectiveChannel channel;
        ASSERT_EQ(0, channel.Init("rr", NULL));
        for (size_t i = 0; i < NCHANS; ++i) {
            brpc::Channel* subchan = new brpc::Channel;
            SetUpChannel(subchan, single_server, short_connection);
            ASSERT_EQ(0, channel.AddChannel(subchan, NULL)) << "i=" << i;
        }
        
        brpc::Controller cntl;
        test::EchoRequest req;
        test::EchoResponse res;
        CallMethod(&channel, &cntl, &req, &res, async);
        EXPECT_EQ(brpc::EREQUEST, cntl.ErrorCode()) << cntl.ErrorText();
        LOG(WARNING) << cntl.ErrorText();
        ASSERT_EQ(1, cntl.sub_count());
        ASSERT_EQ(brpc::EREQUEST, cntl.sub(0)->ErrorCode());
        StopAndJoin();
    }
    
    void TestRPCTimeout(bool single_server, bool async, bool short_connection) {
        std::cout << " *** single=" << single_server
                  << " async=" << async
                  << " short=" << short_connection << std::endl;
        ASSERT_EQ(0, StartAccept(_ep));
        brpc::Channel channel;
        SetUpChannel(&channel, single_server, short_connection);
                
        brpc::Controller cntl;
        test::EchoRequest req;
        test::EchoResponse res;
        req.set_message(__FUNCTION__);
        req.set_sleep_us(70000); // 70ms
        cntl.set_timeout_ms(17);
        butil::Timer tm;
        tm.start();
        CallMethod(&channel, &cntl, &req, &res, async);
        tm.stop();
        EXPECT_EQ(brpc::ERPCTIMEDOUT, cntl.ErrorCode()) << cntl.ErrorText();
        EXPECT_LT(labs(tm.m_elapsed() - cntl.timeout_ms()), 15);
        StopAndJoin();
    }

    void TestRPCTimeoutParallel(
        bool single_server, bool async, bool short_connection) {
        std::cout << " *** single=" << single_server
                  << " async=" << async
                  << " short=" << short_connection << std::endl;
        ASSERT_EQ(0, StartAccept(_ep));
        
        const size_t NCHANS = 8;
        brpc::Channel subchans[NCHANS];
        brpc::ParallelChannel channel;
        for (size_t i = 0; i < NCHANS; ++i) {
            SetUpChannel(&subchans[i], single_server, short_connection);
            ASSERT_EQ(0, channel.AddChannel(
                          &subchans[i], brpc::DOESNT_OWN_CHANNEL,
                          NULL, NULL));
        }
                
        brpc::Controller cntl;
        test::EchoRequest req;
        test::EchoResponse res;
        req.set_message(__FUNCTION__);
        cntl.set_timeout_ms(17);
        req.set_sleep_us(70000); // 70ms
        butil::Timer tm;
        tm.start();
        CallMethod(&channel, &cntl, &req, &res, async);
        tm.stop();
        EXPECT_EQ(brpc::ERPCTIMEDOUT, cntl.ErrorCode()) << cntl.ErrorText();
        EXPECT_EQ(NCHANS, (size_t)cntl.sub_count());
        for (int i = 0; i < cntl.sub_count(); ++i) {
            EXPECT_EQ(ECANCELED, cntl.sub(i)->ErrorCode()) << "i=" << i;
        }
        EXPECT_LT(labs(tm.m_elapsed() - cntl.timeout_ms()), 15);
        StopAndJoin();
    }

    class MakeTheRequestTimeout : public brpc::CallMapper {
    public:
        brpc::SubCall Map(
            int /*channel_index*/,
            const google::protobuf::MethodDescriptor* method,
            const google::protobuf::Message* req_base,
            google::protobuf::Message* response) override {
            test::EchoRequest* req = brpc::Clone<test::EchoRequest>(req_base);
            req->set_sleep_us(70000); // 70ms
            return brpc::SubCall(method, req, response->New(),
                                brpc::DELETE_REQUEST | brpc::DELETE_RESPONSE);
        }
    };

    void TimeoutStillChecksSubChannelsParallel(
        bool single_server, bool async, bool short_connection) {
        std::cout << " *** single=" << single_server
                  << " async=" << async
                  << " short=" << short_connection << std::endl;
        ASSERT_EQ(0, StartAccept(_ep));
        
        const size_t NCHANS = 8;
        brpc::Channel subchans[NCHANS];
        brpc::ParallelChannel channel;
        for (size_t i = 0; i < NCHANS; ++i) {
            SetUpChannel(&subchans[i], single_server, short_connection);
            ASSERT_EQ(0, channel.AddChannel(
                          &subchans[i], brpc::DOESNT_OWN_CHANNEL,
                          ((i % 2) ? new MakeTheRequestTimeout : NULL), NULL));
        }
                
        brpc::Controller cntl;
        test::EchoRequest req;
        test::EchoResponse res;
        req.set_message(__FUNCTION__);
        cntl.set_timeout_ms(30);
        butil::Timer tm;
        tm.start();
        CallMethod(&channel, &cntl, &req, &res, async);
        tm.stop();
        EXPECT_EQ(0, cntl.ErrorCode()) << cntl.ErrorText();
        EXPECT_EQ(NCHANS, (size_t)cntl.sub_count());
        for (int i = 0; i < cntl.sub_count(); ++i) {
            if (i % 2) {
                EXPECT_EQ(ECANCELED, cntl.sub(i)->ErrorCode());
            } else {
                EXPECT_EQ(0, cntl.sub(i)->ErrorCode());
            }
        }
        EXPECT_LT(labs(tm.m_elapsed() - cntl.timeout_ms()), 15);
        StopAndJoin();
    }

    void TestRPCTimeoutSelective(
        bool single_server, bool async, bool short_connection) {
        std::cout << " *** single=" << single_server
                  << " async=" << async
                  << " short=" << short_connection << std::endl;
        ASSERT_EQ(0, StartAccept(_ep));

        const size_t NCHANS = 8;
        brpc::SelectiveChannel channel;
        ASSERT_EQ(0, channel.Init("rr", NULL));
        for (size_t i = 0; i < NCHANS; ++i) {
            brpc::Channel* subchan = new brpc::Channel;
            SetUpChannel(subchan, single_server, short_connection);
            ASSERT_EQ(0, channel.AddChannel(subchan, NULL)) << "i=" << i;
        }

        brpc::Controller cntl;
        test::EchoRequest req;
        test::EchoResponse res;
        req.set_message(__FUNCTION__);
        cntl.set_timeout_ms(17);
        req.set_sleep_us(70000); // 70ms
        butil::Timer tm;
        tm.start();
        CallMethod(&channel, &cntl, &req, &res, async);
        tm.stop();
        EXPECT_EQ(brpc::ERPCTIMEDOUT, cntl.ErrorCode()) << cntl.ErrorText();
        EXPECT_EQ(1, cntl.sub_count());
        EXPECT_EQ(brpc::ERPCTIMEDOUT, cntl.sub(0)->ErrorCode());
        EXPECT_LT(labs(tm.m_elapsed() - cntl.timeout_ms()), 15);
        EXPECT_EQ(-1, cntl.sub(0)->_timeout_ms);
        EXPECT_EQ(17, cntl.sub(0)->_real_timeout_ms);
        StopAndJoin();
    }

    void TestBackupRequestSelective(
        bool single_server, bool async, bool short_connection) {
        std::cout << " *** single=" << single_server
                  << " async=" << async
                  << " short=" << short_connection << std::endl;
        ASSERT_EQ(0, StartAccept(_ep));

        const size_t NCHANS = 8;
        brpc::SelectiveChannel channel;
        ASSERT_EQ(0, channel.Init("rr", NULL));
        for (size_t i = 0; i < NCHANS; ++i) {
            brpc::Channel* subchan = new brpc::Channel;
            SetUpChannel(subchan, single_server, short_connection);
            ASSERT_EQ(0, channel.AddChannel(subchan, NULL)) << "i=" << i;
        }

        brpc::Controller cntl;
        test::EchoRequest req;
        test::EchoResponse res;
        req.set_message(__FUNCTION__);
        cntl.set_backup_request_ms(20);
        cntl.set_timeout_ms(100);
        std::atomic<int> call_cnt(0);
        _svc.SetMockFunc([&call_cnt](google::protobuf::RpcController* cntl_base,
                                     const ::test::EchoRequest*,
                                     ::test::EchoResponse*,
                                     google::protobuf::Closure*) {
            brpc::Controller* cntl = static_cast<brpc::Controller*>(cntl_base);
            int see_cnt = call_cnt.fetch_add(1, std::memory_order_relaxed);
            if (see_cnt == 0) {
                LOG(INFO) << "slow node";
                bthread_usleep(30 * 1000);
            } else {
                LOG(INFO) << "normal node ";
                butil::IOBuf iobuf;
                iobuf.append("123");
                cntl->response_attachment().swap(iobuf);
            }
        });
        butil::Timer tm;
        tm.start();
        CallMethod(&channel, &cntl, &req, &res, async);
        tm.stop();
        EXPECT_FALSE(cntl.Failed());
        EXPECT_EQ(call_cnt.load(std::memory_order_relaxed), 2);
        EXPECT_EQ(cntl.response_attachment().to_string(), "123");
        StopAndJoin();
    }

    void TestBackupRequestSelectiveResponseRace() {
        ASSERT_EQ(0, StartAccept(_ep));

        const size_t NCHANS = 8;
        brpc::SelectiveChannel channel;
        ASSERT_EQ(0, channel.Init("rr", NULL));
        for (size_t i = 0; i < NCHANS; ++i) {
            brpc::Channel* subchan = new brpc::Channel;
            SetUpChannel(subchan, false, false);
            ASSERT_EQ(0, channel.AddChannel(subchan, NULL)) << "i=" << i;
        }

        const int kRounds = 150;
        const int kCodeListSize = 20000;
        std::atomic<int> call_cnt(0);
        _svc.SetMockFunc([&call_cnt](google::protobuf::RpcController*,
                                     const ::test::EchoRequest*,
                                     ::test::EchoResponse* res,
                                     google::protobuf::Closure*) {
            const int seen = call_cnt.fetch_add(1, std::memory_order_relaxed);
            const bool slow = ((seen & 1) == 0);
            if (slow) {
                bthread_usleep(1500);
            }
            res->clear_code_list();
            const int base = slow ? 1000000 : 2000000;
            for (int i = 0; i < kCodeListSize; ++i) {
                res->add_code_list(base + i);
            }
            res->set_message(slow ? "slow" : "fast");
        });

        for (int round = 0; round < kRounds; ++round) {
            brpc::Controller cntl;
            test::EchoRequest req;
            test::EchoResponse res;
            req.set_message(__FUNCTION__);
            cntl.set_backup_request_ms(1);
            cntl.set_timeout_ms(3000);
            CallMethod(&channel, &cntl, &req, &res, true);
            ASSERT_FALSE(cntl.Failed()) << "round=" << round
                                        << " err=" << cntl.ErrorText();
            ASSERT_EQ(kCodeListSize, res.code_list_size()) << "round=" << round;
            ASSERT_TRUE(res.message() == "slow" || res.message() == "fast")
                << "round=" << round;
        }

        EXPECT_EQ(kRounds * 2, call_cnt.load(std::memory_order_relaxed));
        StopAndJoin();
    }
    
    void TestCloseFD(bool single_server, bool async, bool short_connection) {
        std::cout << " *** single=" << single_server
                  << " async=" << async
                  << " short=" << short_connection << std::endl;

        ASSERT_EQ(0, StartAccept(_ep));
        brpc::Channel channel;
        SetUpChannel(&channel, single_server, short_connection);
                
        brpc::Controller cntl;
        test::EchoRequest req;
        test::EchoResponse res;
        req.set_message(__FUNCTION__);
        req.set_close_fd(true);
        CallMethod(&channel, &cntl, &req, &res, async);
        
        EXPECT_EQ(brpc::EEOF, cntl.ErrorCode()) << cntl.ErrorText();
        StopAndJoin();
    }

    void TestCloseFDParallel(bool single_server, bool async, bool short_connection) {
        std::cout << " *** single=" << single_server
                  << " async=" << async
                  << " short=" << short_connection << std::endl;

        ASSERT_EQ(0, StartAccept(_ep));

        const size_t NCHANS = 8;
        brpc::Channel subchans[NCHANS];
        brpc::ParallelChannel channel;
        for (size_t i = 0; i < NCHANS; ++i) {
            SetUpChannel(&subchans[i], single_server, short_connection);
            ASSERT_EQ(0, channel.AddChannel(
                          &subchans[i], brpc::DOESNT_OWN_CHANNEL,
                          NULL, NULL));
        }

        brpc::Controller cntl;
        test::EchoRequest req;
        test::EchoResponse res;
        req.set_message(__FUNCTION__);
        req.set_close_fd(true);
        CallMethod(&channel, &cntl, &req, &res, async);
        
        EXPECT_TRUE(brpc::EEOF == cntl.ErrorCode() ||
                    brpc::ETOOMANYFAILS == cntl.ErrorCode() ||
                    ECONNRESET == cntl.ErrorCode()) << cntl.ErrorText();
        StopAndJoin();
    }

    void TestCloseFDSelective(bool single_server, bool async, bool short_connection) {
        std::cout << " *** single=" << single_server
                  << " async=" << async
                  << " short=" << short_connection << std::endl;

        ASSERT_EQ(0, StartAccept(_ep));

        const size_t NCHANS = 8;
        brpc::SelectiveChannel channel;
        brpc::ChannelOptions options;
        options.max_retry = 0;
        ASSERT_EQ(0, channel.Init("rr", &options));
        for (size_t i = 0; i < NCHANS; ++i) {
            brpc::Channel* subchan = new brpc::Channel;
            SetUpChannel(subchan, single_server, short_connection);
            ASSERT_EQ(0, channel.AddChannel(subchan, NULL)) << "i=" << i;
        }

        brpc::Controller cntl;
        test::EchoRequest req;
        test::EchoResponse res;
        req.set_message(__FUNCTION__);
        req.set_close_fd(true);
        CallMethod(&channel, &cntl, &req, &res, async);
        
        EXPECT_EQ(brpc::EEOF, cntl.ErrorCode()) << cntl.ErrorText();
        ASSERT_EQ(1, cntl.sub_count());
        ASSERT_EQ(brpc::EEOF, cntl.sub(0)->ErrorCode());

        StopAndJoin();
    }
    
    void TestServerFail(bool single_server, bool async, bool short_connection) {
        std::cout << " *** single=" << single_server
                  << " async=" << async
                  << " short=" << short_connection << std::endl;

        ASSERT_EQ(0, StartAccept(_ep));
        brpc::Channel channel;
        SetUpChannel(&channel, single_server, short_connection);
                
        brpc::Controller cntl;
        test::EchoRequest req;
        test::EchoResponse res;
        req.set_message(__FUNCTION__);
        req.set_server_fail(brpc::EINTERNAL);
        CallMethod(&channel, &cntl, &req, &res, async);
        
        EXPECT_EQ(brpc::EINTERNAL, cntl.ErrorCode()) << cntl.ErrorText();
        StopAndJoin();
    }

    void TestServerFailParallel(bool single_server, bool async, bool short_connection) {
        std::cout << " *** single=" << single_server
                  << " async=" << async
                  << " short=" << short_connection << std::endl;

        ASSERT_EQ(0, StartAccept(_ep));

        const size_t NCHANS = 8;
        brpc::Channel subchans[NCHANS];
        brpc::ParallelChannel channel;
        for (size_t i = 0; i < NCHANS; ++i) {
            SetUpChannel(&subchans[i], single_server, short_connection);
            ASSERT_EQ(0, channel.AddChannel(
                          &subchans[i], brpc::DOESNT_OWN_CHANNEL,
                          NULL, NULL));
        }

        brpc::Controller cntl;
        test::EchoRequest req;
        test::EchoResponse res;
        req.set_message(__FUNCTION__);
        req.set_server_fail(brpc::EINTERNAL);
        CallMethod(&channel, &cntl, &req, &res, async);
        
        EXPECT_EQ(brpc::EINTERNAL, cntl.ErrorCode()) << cntl.ErrorText();
        LOG(INFO) << cntl.ErrorText();
        StopAndJoin();
    }

    void TestServerFailSelective(bool single_server, bool async, bool short_connection) {
        std::cout << " *** single=" << single_server
                  << " async=" << async
                  << " short=" << short_connection << std::endl;

        ASSERT_EQ(0, StartAccept(_ep));

        const size_t NCHANS = 5;
        brpc::SelectiveChannel channel;
        ASSERT_EQ(0, channel.Init("rr", NULL));
        for (size_t i = 0; i < NCHANS; ++i) {
            brpc::Channel* subchan = new brpc::Channel;
            SetUpChannel(subchan, single_server, short_connection);
            ASSERT_EQ(0, channel.AddChannel(subchan, NULL)) << "i=" << i;
        }

        brpc::Controller cntl;
        test::EchoRequest req;
        test::EchoResponse res;
        req.set_message(__FUNCTION__);
        req.set_server_fail(brpc::EINTERNAL);
        CallMethod(&channel, &cntl, &req, &res, async);
        
        EXPECT_EQ(brpc::EINTERNAL, cntl.ErrorCode()) << cntl.ErrorText();
        ASSERT_EQ(1, cntl.sub_count());
        ASSERT_EQ(brpc::EINTERNAL, cntl.sub(0)->ErrorCode());

        LOG(INFO) << cntl.ErrorText();
        StopAndJoin();
    }
    
    void TestDestroyChannel(bool single_server, bool short_connection) {
        std::cout << "*** single=" << single_server
                  << ", short=" << short_connection << std::endl;

        ASSERT_EQ(0, StartAccept(_ep));
        brpc::Channel* channel = new brpc::Channel();
        SetUpChannel(channel, single_server, short_connection);
                
        brpc::Controller cntl;
        test::EchoRequest req;
        test::EchoResponse res;
        req.set_message(__FUNCTION__);
        req.set_sleep_us(10000);
        CallMethod(channel, &cntl, &req, &res, true, true/*destroy*/);

        EXPECT_EQ(0, cntl.ErrorCode()) << cntl.ErrorText();
        EXPECT_EQ("received " + std::string(__FUNCTION__), res.message());
        // Sleep to let `_messenger' detect `Socket' being `SetFailed'
        const int64_t start_time = butil::cpuwide_time_us();
        while (_messenger.ConnectionCount() != 0) {
            EXPECT_LT(butil::cpuwide_time_us(), start_time + 100000L/*100ms*/);
            bthread_usleep(1000);
        }

        StopAndJoin();
    }
    
    void TestDestroyChannelParallel(bool single_server, bool short_connection) {
        std::cout << "*** single=" << single_server
                  << ", short=" << short_connection << std::endl;

        const size_t NCHANS = 5;
        ASSERT_EQ(0, StartAccept(_ep));
        brpc::ParallelChannel* channel = new brpc::ParallelChannel;
        for (size_t i = 0; i < NCHANS; ++i) {
            brpc::Channel* subchan = new brpc::Channel();
            SetUpChannel(subchan, single_server, short_connection);
            ASSERT_EQ(0, channel->AddChannel(
                          subchan, brpc::OWNS_CHANNEL, NULL, NULL));
        }
                
        brpc::Controller cntl;
        test::EchoRequest req;
        test::EchoResponse res;
        req.set_sleep_us(10000);
        req.set_message(__FUNCTION__);
        CallMethod(channel, &cntl, &req, &res, true, true/*destroy*/);
        
        EXPECT_EQ(0, cntl.ErrorCode()) << cntl.ErrorText();
        EXPECT_EQ("received " + std::string(__FUNCTION__), res.message());
        // Sleep to let `_messenger' detect `Socket' being `SetFailed'
        const int64_t start_time = butil::cpuwide_time_us();
        while (_messenger.ConnectionCount() != 0) {
            EXPECT_LT(butil::cpuwide_time_us(), start_time + 100000L/*100ms*/);
            bthread_usleep(1000);
        }
        StopAndJoin();
    }

    void TestDestroyChannelSelective(bool single_server, bool short_connection) {
        std::cout << "*** single=" << single_server
                  << ", short=" << short_connection << std::endl;

        const size_t NCHANS = 5;
        ASSERT_EQ(0, StartAccept(_ep));
        brpc::SelectiveChannel* channel = new brpc::SelectiveChannel;
        ASSERT_EQ(0, channel->Init("rr", NULL));
        for (size_t i = 0; i < NCHANS; ++i) {
            brpc::Channel* subchan = new brpc::Channel();
            SetUpChannel(subchan, single_server, short_connection);
            ASSERT_EQ(0, channel->AddChannel(subchan, NULL));
        }
                
        brpc::Controller cntl;
        test::EchoRequest req;
        test::EchoResponse res;
        req.set_sleep_us(10000);
        req.set_message(__FUNCTION__);
        CallMethod(channel, &cntl, &req, &res, true, true/*destroy*/);
        
        EXPECT_EQ(0, cntl.ErrorCode()) << cntl.ErrorText();
        EXPECT_EQ("received " + std::string(__FUNCTION__), res.message());
        ASSERT_EQ(_ep, cntl.remote_side());
        ASSERT_EQ(1, cntl.sub_count());
        ASSERT_EQ(0, cntl.sub(0)->ErrorCode());

        // Sleep to let `_messenger' detect `Socket' being `SetFailed'
        const int64_t start_time = butil::cpuwide_time_us();
        while (_messenger.ConnectionCount() != 0) {
            EXPECT_LT(butil::cpuwide_time_us(), start_time + 100000L/*100ms*/);
            bthread_usleep(1000);
        }
        StopAndJoin();
    }

    void RPCThread(brpc::ChannelBase* channel, bool async) {
        brpc::Controller cntl;
        test::EchoRequest req;
        test::EchoResponse res;
        req.set_message(__FUNCTION__);
        CallMethod(channel, &cntl, &req, &res, async);

        ASSERT_EQ(0, cntl.ErrorCode()) << cntl.ErrorText();
        EXPECT_EQ("received " + std::string(__FUNCTION__), res.message());
    }

    void RPCThread(brpc::ChannelBase* channel, bool async, int count) {
        brpc::Controller cntl;
        for (int i = 0; i < count; ++i) {
            test::EchoRequest req;
            test::EchoResponse res;
            req.set_message(__FUNCTION__);
            CallMethod(channel, &cntl, &req, &res, async);

            ASSERT_EQ(0, cntl.ErrorCode()) << cntl.ErrorText();
            ASSERT_EQ("received " + std::string(__FUNCTION__), res.message());
            cntl.Reset();
        }
    }

    void RPCThread(bool single_server, bool async, bool short_connection,
                   const brpc::Authenticator* auth, int count) {
        brpc::Channel channel;
        SetUpChannel(&channel, single_server, short_connection, auth);
        brpc::Controller cntl;
        for (int i = 0; i < count; ++i) {
            test::EchoRequest req;
            test::EchoResponse res;
            req.set_message(__FUNCTION__);
            CallMethod(&channel, &cntl, &req, &res, async);

            ASSERT_EQ(0, cntl.ErrorCode()) << cntl.ErrorText();
            ASSERT_EQ("received " + std::string(__FUNCTION__), res.message());
            cntl.Reset();
        }
    }

    void TestAuthentication(bool single_server, 
                            bool async, bool short_connection) {
        std::cout << " *** single=" << single_server
                  << " async=" << async
                  << " short=" << short_connection << std::endl;

        ASSERT_EQ(0, StartAccept(_ep));
        MyAuthenticator auth;
        brpc::Channel channel;
        SetUpChannel(&channel, single_server, short_connection, &auth);

        const int NUM = 10;
        pthread_t tids[NUM];
        for (int i = 0; i < NUM; ++i) {
            google::protobuf::Closure* thrd_func = 
                brpc::NewCallback(
                    this, &ChannelTest::RPCThread, (brpc::ChannelBase*)&channel, async);
            EXPECT_EQ(0, pthread_create(&tids[i], NULL,
                                        RunClosure, thrd_func));
        }
        for (int i = 0; i < NUM; ++i) {
            pthread_join(tids[i], NULL);
        }
        
        if (short_connection) {
            EXPECT_EQ(NUM, auth.count.load());
        } else {
            EXPECT_EQ(1, auth.count.load());
        }
        StopAndJoin();
    }

    void TestAuthenticationParallel(bool single_server, 
                                    bool async, bool short_connection) {
        std::cout << " *** single=" << single_server
                  << " async=" << async
                  << " short=" << short_connection << std::endl;

        ASSERT_EQ(0, StartAccept(_ep));
        MyAuthenticator auth;

        const int NCHANS = 5;
        brpc::Channel subchans[NCHANS];
        brpc::ParallelChannel channel;
        for (int i = 0; i < NCHANS; ++i) {
            SetUpChannel(&subchans[i], single_server, short_connection, &auth);
            ASSERT_EQ(0, channel.AddChannel(
                          &subchans[i], brpc::DOESNT_OWN_CHANNEL,
                          NULL, NULL));
        }
        
        const int NUM = 10;
        pthread_t tids[NUM];
        for (int i = 0; i < NUM; ++i) {
            google::protobuf::Closure* thrd_func = 
                brpc::NewCallback(
                    this, &ChannelTest::RPCThread, (brpc::ChannelBase*)&channel, async);
            EXPECT_EQ(0, pthread_create(&tids[i], NULL,
                                        RunClosure, thrd_func));
        }
        for (int i = 0; i < NUM; ++i) {
            pthread_join(tids[i], NULL);
        }
        
        if (short_connection) {
            EXPECT_EQ(NUM * NCHANS, auth.count.load());
        } else {
            EXPECT_EQ(1, auth.count.load());
        }
        StopAndJoin();
    }

    void TestAuthenticationSelective(bool single_server, 
                                    bool async, bool short_connection) {
        std::cout << " *** single=" << single_server
                  << " async=" << async
                  << " short=" << short_connection << std::endl;

        ASSERT_EQ(0, StartAccept(_ep));
        MyAuthenticator auth;

        const size_t NCHANS = 5;
        brpc::SelectiveChannel channel;
        ASSERT_EQ(0, channel.Init("rr", NULL));
        for (size_t i = 0; i < NCHANS; ++i) {
            brpc::Channel* subchan = new brpc::Channel;
            SetUpChannel(subchan, single_server, short_connection, &auth);
            ASSERT_EQ(0, channel.AddChannel(subchan, NULL)) << "i=" << i;
        }
        
        const int NUM = 10;
        pthread_t tids[NUM];
        for (int i = 0; i < NUM; ++i) {
            google::protobuf::Closure* thrd_func = 
                brpc::NewCallback(
                    this, &ChannelTest::RPCThread, (brpc::ChannelBase*)&channel, async);
            EXPECT_EQ(0, pthread_create(&tids[i], NULL,
                                        RunClosure, thrd_func));
        }
        for (int i = 0; i < NUM; ++i) {
            pthread_join(tids[i], NULL);
        }
        
        if (short_connection) {
            EXPECT_EQ(NUM, auth.count.load());
        } else {
            EXPECT_EQ(1, auth.count.load());
        }
        StopAndJoin();
    }
    
    void TestRetry(bool single_server, bool async, bool short_connection) {
        std::cout << " *** single=" << single_server
                  << " async=" << async
                  << " short=" << short_connection << std::endl;

        ASSERT_EQ(0, StartAccept(_ep));
        brpc::Channel channel;
        SetUpChannel(&channel, single_server, short_connection);

        const int RETRY_NUM = 3;
        test::EchoRequest req;
        test::EchoResponse res;
        brpc::Controller cntl;
        req.set_message(__FUNCTION__);

        // No retry when timeout
        cntl.set_max_retry(RETRY_NUM);
        cntl.set_timeout_ms(10);  // 10ms
        req.set_sleep_us(70000); // 70ms
        CallMethod(&channel, &cntl, &req, &res, async);
        EXPECT_EQ(brpc::ERPCTIMEDOUT, cntl.ErrorCode()) << cntl.ErrorText();
        EXPECT_EQ(0, cntl.retried_count());
        bthread_usleep(100000);  // wait for the sleep task to finish

        // Retry when connection broken
        cntl.Reset();
        cntl.set_max_retry(RETRY_NUM);
        _close_fd_once = true;
        req.set_sleep_us(0);
        CallMethod(&channel, &cntl, &req, &res, async);

        if (short_connection) {
            // Always succeed
            EXPECT_EQ(0, cntl.ErrorCode()) << cntl.ErrorText();
            EXPECT_EQ(1, cntl.retried_count());

            const int64_t start_time = butil::cpuwide_time_us();
            while (_messenger.ConnectionCount() != 0) {
                EXPECT_LT(butil::cpuwide_time_us(), start_time + 100000L/*100ms*/);
                bthread_usleep(1000);
            }
        } else {
            // May fail if health checker can't revive in time
            if (cntl.Failed()) {
                EXPECT_EQ(EHOSTDOWN, cntl.ErrorCode()) << single_server << ", " << async;
                EXPECT_EQ(RETRY_NUM, cntl.retried_count());
            } else {
                EXPECT_TRUE(cntl.retried_count() > 0);
            }
        }   
        StopAndJoin();
        bthread_usleep(100000);  // wait for stop
        
        // Retry when connection failed
        cntl.Reset();
        cntl.set_max_retry(RETRY_NUM);
        CallMethod(&channel, &cntl, &req, &res, async);
        EXPECT_EQ(EHOSTDOWN, cntl.ErrorCode());
        EXPECT_EQ(RETRY_NUM, cntl.retried_count());
    }

    void TestRetryOtherServer(bool async, bool short_connection) {
        ASSERT_EQ(0, StartAccept(_ep));

        brpc::Channel channel;
        brpc::ChannelOptions opt;
        opt.timeout_ms = 1000;
        if (short_connection) {
            opt.connection_type = brpc::CONNECTION_TYPE_SHORT;
        }
        butil::TempFile server_list;                                        
        EXPECT_EQ(0, server_list.save_format(
                      "127.0.0.1:100\n"
                      "127.0.0.1:200\n"
                      "%s", endpoint2str(_ep).c_str()));
        std::string naming_url = std::string("fIle://")
            + server_list.fname();
        EXPECT_EQ(0, channel.Init(naming_url.c_str(), "RR", &opt)); 

        const int RETRY_NUM = 3;
        test::EchoRequest req;
        test::EchoResponse res;
        brpc::Controller cntl;
        req.set_message(__FUNCTION__);
        cntl.set_max_retry(RETRY_NUM);
        CallMethod(&channel, &cntl, &req, &res, async);

        EXPECT_EQ(0, cntl.ErrorCode()) << async << ", " << short_connection;
        StopAndJoin();
    }

    struct TestRetryBackoffInfo {
        TestRetryBackoffInfo(ChannelTest* channel_test_param,
                             bool async_param,
                             bool short_connection_param,
                             bool fixed_backoff_param)
            : channel_test(channel_test_param)
            , async(async_param)
            , short_connection(short_connection_param)
            , fixed_backoff(fixed_backoff_param) {}

        ChannelTest* channel_test;
        int async;
        int short_connection;
        int fixed_backoff;
    };

    static void* TestRetryBackoffBthread(void* void_args) {
        auto args = static_cast<TestRetryBackoffInfo*>(void_args);
        args->channel_test->TestRetryBackoff(args->async, args->short_connection,
                                             args->fixed_backoff, false);
        return NULL;
    }

    void TestRetryBackoff(bool async, bool short_connection, bool fixed_backoff,
                          bool retry_backoff_in_pthread) {
        ASSERT_EQ(0, StartAccept(_ep));

        const int32_t backoff_time_ms = 100;
        const int32_t no_backoff_remaining_rpc_time_ms = 100;
        std::unique_ptr<brpc::RetryPolicy> retry_ptr;
        if (fixed_backoff) {
            retry_ptr.reset(
                    new brpc::RpcRetryPolicyWithFixedBackoff(backoff_time_ms,
                                                             no_backoff_remaining_rpc_time_ms,
                                                             retry_backoff_in_pthread));
        } else {
            retry_ptr.reset(
                    new brpc::RpcRetryPolicyWithJitteredBackoff(backoff_time_ms,
                                                                backoff_time_ms + 20,
                                                                no_backoff_remaining_rpc_time_ms,
                                                                retry_backoff_in_pthread));
        }

        brpc::Channel channel;
        brpc::ChannelOptions opt;
        opt.timeout_ms = 1000;
        opt.retry_policy = retry_ptr.get();
        if (short_connection) {
            opt.connection_type = brpc::CONNECTION_TYPE_SHORT;
        }
        butil::TempFile server_list;
        EXPECT_EQ(0, server_list.save_format(
            "127.0.0.1:100\n"
            "127.0.0.1:200\n"
            "%s", endpoint2str(_ep).c_str()));
        std::string naming_url = std::string("fIle://")
            + server_list.fname();
        EXPECT_EQ(0, channel.Init(naming_url.c_str(), "RR", &opt));

        const int RETRY_NUM = 3;
        test::EchoRequest req;
        test::EchoResponse res;
        brpc::Controller cntl;
        req.set_message(__FUNCTION__);
        cntl.set_max_retry(RETRY_NUM);
        CallMethod(&channel, &cntl, &req, &res, async);
        if (cntl.retried_count() > 0) {
            EXPECT_GT(cntl.latency_us(), ((int64_t)backoff_time_ms * 1000) * cntl.retried_count())
                << "latency_us=" << cntl.latency_us() << " retried_count=" << cntl.retried_count()
                << " enable_retry_backoff_in_pthread=" << retry_backoff_in_pthread;
        }
        EXPECT_EQ(0, cntl.ErrorCode()) << async << ", " << short_connection;
        StopAndJoin();
    }

    void TestBackupRequest(bool single_server, bool async,
                           bool short_connection) {
        std::cout << " *** single=" << single_server
                  << " async=" << async
                  << " short=" << short_connection << std::endl;

        ASSERT_EQ(0, StartAccept(_ep));
        brpc::Channel channel;
        SetUpChannel(&channel, single_server, short_connection);

        const int RETRY_NUM = 1;
        test::EchoRequest req;
        test::EchoResponse res;
        brpc::Controller cntl;
        req.set_message(__FUNCTION__);

        cntl.set_max_retry(RETRY_NUM);
        cntl.set_backup_request_ms(10);  // 10ms
        cntl.set_timeout_ms(100);  // 10ms
        req.set_sleep_us(50000); // 100ms
        CallMethod(&channel, &cntl, &req, &res, async);
        ASSERT_EQ(0, cntl.ErrorCode()) << cntl.ErrorText();
        ASSERT_TRUE(cntl.has_backup_request());
        ASSERT_EQ(RETRY_NUM, cntl.retried_count());
        bthread_usleep(70000);  // wait for the sleep task to finish

        if (short_connection) {
            // Sleep to let `_messenger' detect `Socket' being `SetFailed'
            const int64_t start_time = butil::cpuwide_time_us();
            while (_messenger.ConnectionCount() != 0) {
                EXPECT_LT(butil::cpuwide_time_us(), start_time + 100000L/*100ms*/);
                bthread_usleep(1000);
            }
        } else {
            EXPECT_GE(1ul, _messenger.ConnectionCount());
        }
        StopAndJoin();
    }

    class BackupRequestPolicyImpl : public brpc::BackupRequestPolicy {
    public:
        int32_t GetBackupRequestMs(const brpc::Controller*) const override {
            return 10;
        }

        // Return true if the backup request should be sent.
        bool DoBackup(const brpc::Controller*) const override {
            return backup;
        }

        void OnRPCEnd(const brpc::Controller*) override {}

        bool backup{true};

    };

    void TestBackupRequestPolicy(bool single_server, bool async,
                                 bool short_connection) {
        ASSERT_EQ(0, StartAccept(_ep));
        for (int i = 0; i < 2; ++i) {
            bool backup = i == 0;
            std::cout << " *** single=" << single_server
                      << " async=" << async
                      << " short=" << short_connection
                      << " backup=" << backup
                      << std::endl;

            brpc::Channel channel;
            SetUpChannel(&channel, single_server, short_connection, NULL, "", true);

            const int RETRY_NUM = 1;
            test::EchoRequest req;
            test::EchoResponse res;
            brpc::Controller cntl;
            req.set_message(__FUNCTION__);

            _backup_request_policy.backup = backup;
            cntl.set_max_retry(RETRY_NUM);
            cntl.set_timeout_ms(100);  // 100ms
            req.set_sleep_us(50000); // 50ms
            CallMethod(&channel, &cntl, &req, &res, async);
            ASSERT_EQ(0, cntl.ErrorCode()) << cntl.ErrorText();
            ASSERT_EQ(backup, cntl.has_backup_request());
            ASSERT_EQ(backup ? RETRY_NUM : 0, cntl.retried_count());
            bthread_usleep(70000);  // wait for the sleep task to finish

            if (short_connection) {
                // Sleep to let `_messenger' detect `Socket' being `SetFailed'
                const int64_t start_time = butil::cpuwide_time_us();
                while (_messenger.ConnectionCount() != 0) {
                    ASSERT_LT(butil::cpuwide_time_us(), start_time + 100000L/*100ms*/);
                    bthread_usleep(1000);
                }
            } else {
                ASSERT_GE(1ul, _messenger.ConnectionCount());
            }
        }

        StopAndJoin();
    }

    butil::EndPoint _ep;
    butil::TempFile _server_list;                                        
    std::string _naming_url;
    
    brpc::Acceptor _messenger;
    // Dummy server for `Server::AddError'
    brpc::Server _dummy;
    std::string _mock_fail_str;

    bool _close_fd_once;
    
    MyEchoService _svc;
    BackupRequestPolicyImpl _backup_request_policy;
};

class MyShared : public brpc::SharedObject {
public:
    MyShared() { ++ nctor; }
    MyShared(const MyShared&) : brpc::SharedObject() { ++ nctor; }
    ~MyShared() override { ++ ndtor; }

    static int nctor;
    static int ndtor;
};
int MyShared::nctor = 0;
int MyShared::ndtor = 0;

TEST_F(ChannelTest, intrusive_ptr_sanity) {
    MyShared::nctor = 0;
    MyShared::ndtor = 0;
    {
        MyShared* s1 = new MyShared;
        ASSERT_EQ(0, s1->ref_count());
        butil::intrusive_ptr<MyShared> p1 = s1;
        ASSERT_EQ(1, p1->ref_count());
        {
            butil::intrusive_ptr<MyShared> p2 = s1;
            ASSERT_EQ(2, p2->ref_count());
            ASSERT_EQ(2, p1->ref_count());
        }
        ASSERT_EQ(1, p1->ref_count());
    }
    ASSERT_EQ(1, MyShared::nctor);
    ASSERT_EQ(1, MyShared::ndtor);
}

TEST_F(ChannelTest, init_as_single_server) {
    {
        brpc::Channel channel;
        ASSERT_EQ(-1, channel.Init("127.0.0.1:12345:asdf", NULL));
        ASSERT_EQ(-1, channel.Init("127.0.0.1:99999", NULL)); 
        ASSERT_EQ(0, channel.Init("127.0.0.1:8888", NULL));
    }
    {
        brpc::Channel channel;
        ASSERT_EQ(-1, channel.Init("127.0.0.1asdf", 12345, NULL));
        ASSERT_EQ(-1, channel.Init("127.0.0.1", 99999, NULL));
        ASSERT_EQ(0, channel.Init("127.0.0.1", 8888, NULL));
    }

    butil::EndPoint ep;
    brpc::Channel channel;
    ASSERT_EQ(0, str2endpoint("127.0.0.1:8888", &ep));
    ASSERT_EQ(0, channel.Init(ep, NULL));
    ASSERT_TRUE(channel.SingleServer());
    ASSERT_EQ(ep, channel._server_address);

    brpc::SocketId id;
    ASSERT_EQ(0, brpc::SocketMapFind(brpc::SocketMapKey(ep), &id));
    ASSERT_EQ(id, channel._server_id);

    const int NUM = 10;
    brpc::Channel channels[NUM];
    for (int i = 0; i < 10; ++i) {
        ASSERT_EQ(0, channels[i].Init(ep, NULL));
        // Share the same server socket
        ASSERT_EQ(id, channels[i]._server_id);
    }
}

TEST_F(ChannelTest, init_using_unknown_naming_service) {
    brpc::Channel channel;
    ASSERT_EQ(-1, channel.Init("unknown://unknown", "unknown", NULL));
}

TEST_F(ChannelTest, init_using_unexist_fns) {
    brpc::Channel channel;
    ASSERT_EQ(-1, channel.Init("fiLe://no_such_file", "rr", NULL));
}

TEST_F(ChannelTest, init_using_empty_fns) {
    brpc::ChannelOptions opt;
    opt.succeed_without_server = false;
    brpc::Channel channel;
    butil::TempFile server_list;
    ASSERT_EQ(0, server_list.save(""));
    std::string naming_url = std::string("file://") + server_list.fname();
    // empty file list results in error.
    ASSERT_EQ(-1, channel.Init(naming_url.c_str(), "rr", &opt));

    ASSERT_EQ(0, server_list.save("blahblah"));
    // No valid address.
    ASSERT_EQ(-1, channel.Init(naming_url.c_str(), "rr", NULL));
}

TEST_F(ChannelTest, init_using_empty_lns) {
    brpc::ChannelOptions opt;
    opt.succeed_without_server = false;
    brpc::Channel channel;
    ASSERT_EQ(-1, channel.Init("list:// ", "rr", &opt));
    ASSERT_EQ(-1, channel.Init("list://", "rr", &opt));
    ASSERT_EQ(-1, channel.Init("list://blahblah", "rr", &opt)); 
}

TEST_F(ChannelTest, init_using_naming_service) {
    brpc::Channel* channel = new brpc::Channel();
    butil::TempFile server_list;
    ASSERT_EQ(0, server_list.save("127.0.0.1:8888"));
    std::string naming_url = std::string("filE://") + server_list.fname();
    // Rr are intended to test case-insensitivity.
    ASSERT_EQ(0, channel->Init(naming_url.c_str(), "Rr", NULL));
    ASSERT_FALSE(channel->SingleServer());

    brpc::LoadBalancerWithNaming* lb =
        dynamic_cast<brpc::LoadBalancerWithNaming*>(channel->_lb.get());
    ASSERT_TRUE(lb != NULL);
    brpc::NamingServiceThread* ns = lb->_nsthread_ptr.get();

    {
        const int NUM = 10;
        brpc::Channel channels[NUM];
        for (int i = 0; i < NUM; ++i) {
            // Share the same naming thread
            ASSERT_EQ(0, channels[i].Init(naming_url.c_str(), "rr", NULL));
            brpc::LoadBalancerWithNaming* lb2 =
                dynamic_cast<brpc::LoadBalancerWithNaming*>(channels[i]._lb.get());
            ASSERT_TRUE(lb2 != NULL);
            ASSERT_EQ(ns, lb2->_nsthread_ptr.get());
        }
    }

    // `lb' should be valid even if `channel' has destroyed
    // since we hold another reference to it
    butil::intrusive_ptr<brpc::SharedLoadBalancer>
        another_ctx = channel->_lb;
    delete channel;
    ASSERT_EQ(lb, another_ctx.get());
    ASSERT_EQ(1, another_ctx->_nref.load());
    // `lb' should be destroyed after
}

TEST_F(ChannelTest, parse_hostname) {
    brpc::ChannelOptions opt;
    opt.succeed_without_server = false;
    opt.protocol = brpc::PROTOCOL_HTTP;
    brpc::Channel channel;

    ASSERT_EQ(-1, channel.Init("", 8888, &opt));
    ASSERT_EQ("", channel._service_name);
    ASSERT_EQ(-1, channel.Init("", &opt));
    ASSERT_EQ("", channel._service_name);

    ASSERT_EQ(0, channel.Init("http://127.0.0.1", 8888, &opt));
    ASSERT_EQ("127.0.0.1:8888", channel._service_name);
    ASSERT_EQ(0, channel.Init("http://127.0.0.1:8888", &opt));
    ASSERT_EQ("127.0.0.1:8888", channel._service_name);

    ASSERT_EQ(0, channel.Init("localhost", 8888, &opt));
    ASSERT_EQ("localhost:8888", channel._service_name);
    ASSERT_EQ(0, channel.Init("localhost:8888", &opt));
    ASSERT_EQ("localhost:8888", channel._service_name);

    ASSERT_EQ(0, channel.Init("http://www.baidu.com", &opt));
    ASSERT_EQ("www.baidu.com", channel._service_name);
    ASSERT_EQ(0, channel.Init("http://www.baidu.com:80", &opt));
    ASSERT_EQ("www.baidu.com:80", channel._service_name);
    ASSERT_EQ(0, channel.Init("http://www.baidu.com", 80, &opt));
    ASSERT_EQ("www.baidu.com:80", channel._service_name);
    ASSERT_EQ(0, channel.Init("http://www.baidu.com:8888", &opt));
    ASSERT_EQ("www.baidu.com:8888", channel._service_name);
    ASSERT_EQ(0, channel.Init("http://www.baidu.com", 8888, &opt));
    ASSERT_EQ("www.baidu.com:8888", channel._service_name);
    ASSERT_EQ(0, channel.Init("http://www.baidu.com", "rr", &opt));
    ASSERT_EQ("www.baidu.com", channel._service_name);
    ASSERT_EQ(0, channel.Init("http://www.baidu.com:80", "rr", &opt));
    ASSERT_EQ("www.baidu.com:80", channel._service_name);
    ASSERT_EQ(0, channel.Init("http://www.baidu.com:8888", "rr", &opt));
    ASSERT_EQ("www.baidu.com:8888", channel._service_name);

    ASSERT_EQ(0, channel.Init("https://www.baidu.com", &opt));
    ASSERT_EQ("www.baidu.com", channel._service_name);
    ASSERT_EQ(0, channel.Init("https://www.baidu.com:443", &opt));
    ASSERT_EQ("www.baidu.com:443", channel._service_name);
    ASSERT_EQ(0, channel.Init("https://www.baidu.com", 443, &opt));
    ASSERT_EQ("www.baidu.com:443", channel._service_name);
    ASSERT_EQ(0, channel.Init("https://www.baidu.com:1443", &opt));
    ASSERT_EQ("www.baidu.com:1443", channel._service_name);
    ASSERT_EQ(0, channel.Init("https://www.baidu.com", 1443, &opt));
    ASSERT_EQ("www.baidu.com:1443", channel._service_name);
    ASSERT_EQ(0, channel.Init("https://www.baidu.com", "rr", &opt));
    ASSERT_EQ("www.baidu.com", channel._service_name);
    ASSERT_EQ(0, channel.Init("https://www.baidu.com:443", "rr", &opt));
    ASSERT_EQ("www.baidu.com:443", channel._service_name);
    ASSERT_EQ(0, channel.Init("https://www.baidu.com:1443", "rr", &opt));
    ASSERT_EQ("www.baidu.com:1443", channel._service_name);

    const char *address_list[] =  {
        "10.127.0.1:1234",
        "10.128.0.1:1234 enable",
        "10.129.0.1:1234",
        "localhost:1234",
        "www.baidu.com:1234"
    };
    butil::TempFile tmp_file;
    {
        FILE* fp = fopen(tmp_file.fname(), "w");
        for (size_t i = 0; i < ARRAY_SIZE(address_list); ++i) {
            ASSERT_TRUE(fprintf(fp, "%s\n", address_list[i]));
        }
        fclose(fp);
    }
    brpc::Channel ns_channel;
    std::string ns = std::string("file://") + tmp_file.fname();
    ASSERT_EQ(0, ns_channel.Init(ns.c_str(), "rr", &opt));
    ASSERT_EQ(tmp_file.fname(), ns_channel._service_name);
}

TEST_F(ChannelTest, connection_failed) {
    for (int i = 0; i <= 1; ++i) { // Flag SingleServer 
        for (int j = 0; j <= 1; ++j) { // Flag Asynchronous
            for (int k = 0; k <=1; ++k) { // Flag ShortConnection
                TestConnectionFailed(i, j, k);
            }
        }
    }
}

TEST_F(ChannelTest, empty_parallel_channel) {
    brpc::ParallelChannel channel;

    brpc::Controller cntl;
    test::EchoRequest req;
    test::EchoResponse res;
    req.set_message(__FUNCTION__);
    CallMethod(&channel, &cntl, &req, &res, false);
    EXPECT_EQ(EPERM, cntl.ErrorCode()) << cntl.ErrorText();
}

TEST_F(ChannelTest, uninitialized_selective_channel) {
    brpc::SelectiveChannel channel;
    test::EchoRequest req;
    test::EchoResponse res;
    req.set_message(__FUNCTION__);

    brpc::Controller sync_cntl;
    ::test::EchoService::Stub(&channel).Echo(&sync_cntl, &req, &res, NULL);
    EXPECT_EQ(EINVAL, sync_cntl.ErrorCode()) << sync_cntl.ErrorText();

    brpc::Controller async_cntl;
    bool done_called = false;
    google::protobuf::Closure* done =
        brpc::NewCallback(&MarkCalled, &done_called);
    ::test::EchoService::Stub(&channel).Echo(&async_cntl, &req, &res, done);
    EXPECT_TRUE(done_called);
    EXPECT_EQ(EINVAL, async_cntl.ErrorCode()) << async_cntl.ErrorText();
}

TEST_F(ChannelTest, empty_selective_channel) {
    brpc::SelectiveChannel channel;
    ASSERT_EQ(0, channel.Init("rr", NULL));

    brpc::Controller cntl;
    test::EchoRequest req;
    test::EchoResponse res;
    req.set_message(__FUNCTION__);
    CallMethod(&channel, &cntl, &req, &res, false);
    EXPECT_EQ(ENODATA, cntl.ErrorCode()) << cntl.ErrorText();
}

class BadCall : public brpc::CallMapper {
    brpc::SubCall Map(int,
                     const google::protobuf::MethodDescriptor*,
                     const google::protobuf::Message*,
                     google::protobuf::Message*) override {
        return brpc::SubCall::Bad();
    }
};

TEST_F(ChannelTest, returns_bad_parallel) {
    const size_t NCHANS = 5;
    brpc::ParallelChannel channel;
    for (size_t i = 0; i < NCHANS; ++i) {
        brpc::Channel* subchan = new brpc::Channel();
        SetUpChannel(subchan, true, false);
        ASSERT_EQ(0, channel.AddChannel(
                      subchan, brpc::OWNS_CHANNEL, new BadCall, NULL));
    }
                
    brpc::Controller cntl;
    test::EchoRequest req;
    test::EchoResponse res;
    req.set_message(__FUNCTION__);
    CallMethod(&channel, &cntl, &req, &res, false);
    EXPECT_EQ(brpc::EREQUEST, cntl.ErrorCode()) << cntl.ErrorText();
}

class SkipCall : public brpc::CallMapper {
    brpc::SubCall Map(int,
                     const google::protobuf::MethodDescriptor*,
                     const google::protobuf::Message*,
                     google::protobuf::Message*) override {
        return brpc::SubCall::Skip();
    }
};

TEST_F(ChannelTest, skip_all_channels) {
    const size_t NCHANS = 5;
    brpc::ParallelChannel channel;
    for (size_t i = 0; i < NCHANS; ++i) {
        brpc::Channel* subchan = new brpc::Channel();
        SetUpChannel(subchan, true, false);
        ASSERT_EQ(0, channel.AddChannel(
                      subchan, brpc::OWNS_CHANNEL, new SkipCall, NULL));
    }
                
    brpc::Controller cntl;
    test::EchoRequest req;
    test::EchoResponse res;
    req.set_message(__FUNCTION__);
    CallMethod(&channel, &cntl, &req, &res, false);
        
    EXPECT_EQ(ECANCELED, cntl.ErrorCode()) << cntl.ErrorText();
    EXPECT_EQ((int)NCHANS, cntl.sub_count());
    for (int i = 0; i < cntl.sub_count(); ++i) {
        EXPECT_TRUE(NULL == cntl.sub(i)) << "i=" << i;
    }
}

static const std::string ECHO_HTTP_HEADER = "echo-http-header";

class EchoHttpHeader : public brpc::CallMapper {
public:
    brpc::SubCall Map(int channel_index, int channel_count,
                      const google::protobuf::MethodDescriptor* method,
                      const google::protobuf::Message* request,
                      google::protobuf::Message* response) override {
        return brpc::SubCall(method, request, response->New(), brpc::DELETE_RESPONSE);
    }

    void MapController(int channel_index, int,
                       const brpc::Controller* main_cntl,
                       brpc::Controller* sub_cntl) override {
        sub_cntl->http_request().SetHeader(ECHO_HTTP_HEADER, std::to_string(channel_index));
    }
};

TEST_F(ChannelTest, http_header_parallel_channels) {
    brpc::Server server;
    MyEchoService service;
    ASSERT_EQ(0, server.AddService(&service, brpc::SERVER_DOESNT_OWN_SERVICE));
    brpc::ServerOptions opt;
    ASSERT_EQ(0, server.Start(_ep, &opt));

    const size_t NCHANS = 5;
    brpc::ParallelChannel channel;
    for (size_t i = 0; i < NCHANS; ++i) {
        brpc::Channel* sub_chan = new brpc::Channel();
        SetUpChannel(sub_chan, true, false, NULL, "", false, brpc::PROTOCOL_HTTP);
        ASSERT_EQ(0, channel.AddChannel(sub_chan, brpc::OWNS_CHANNEL, new EchoHttpHeader, NULL));
    }

    brpc::Controller cntl;
    test::EchoRequest req;
    test::EchoResponse res;
    req.set_message(__FUNCTION__);
    *req.mutable_http_header() = ECHO_HTTP_HEADER;
    CallMethod(&channel, &cntl, &req, &res, false);

    ASSERT_FALSE(cntl.Failed()) << cntl.ErrorText();
    ASSERT_EQ((int)NCHANS, cntl.sub_count());
    for (int i = 0; i < cntl.sub_count(); ++i) {
        const brpc::Controller* sub_cntl = cntl.sub(i);
        ASSERT_TRUE(NULL != sub_cntl) << "i=" << i;
        ASSERT_EQ(std::to_string(i), *sub_cntl->http_response().GetHeader(ECHO_HTTP_HEADER));
    }
}

TEST_F(ChannelTest, connection_failed_parallel) {
    for (int i = 0; i <= 1; ++i) { // Flag SingleServer 
        for (int j = 0; j <= 1; ++j) { // Flag Asynchronous
            for (int k = 0; k <=1; ++k) { // Flag ShortConnection
                TestConnectionFailedParallel(i, j, k);
            }
        }
    }
}

TEST_F(ChannelTest, connection_failed_selective) {
    for (int i = 0; i <= 1; ++i) { // Flag SingleServer 
        for (int j = 0; j <= 1; ++j) { // Flag Asynchronous
            for (int k = 0; k <=1; ++k) { // Flag ShortConnection
                TestConnectionFailedSelective(i, j, k);
            }
        }
    }
}

TEST_F(ChannelTest, success) {
    for (int i = 0; i <= 1; ++i) { // Flag SingleServer 
        for (int j = 0; j <= 1; ++j) { // Flag Asynchronous
            for (int k = 0; k <=1; ++k) { // Flag ShortConnection
                TestSuccess(i, j, k);
            }
        }
    }
}

TEST_F(ChannelTest, success_parallel) {
    for (int i = 0; i <= 1; ++i) { // Flag SingleServer 
        for (int j = 0; j <= 1; ++j) { // Flag Asynchronous
            for (int k = 0; k <=1; ++k) { // Flag ShortConnection
                TestSuccessParallel(i, j, k);
            }
        }
    }
}

TEST_F(ChannelTest, success_duplicated_parallel) {
    for (int i = 0; i <= 1; ++i) { // Flag SingleServer
        for (int j = 0; j <= 1; ++j) { // Flag Asynchronous
            for (int k = 0; k <=1; ++k) { // Flag ShortConnection
                TestSuccessDuplicatedParallel(i, j, k);
            }
        }
    }
}

TEST_F(ChannelTest, success_selective) {
    for (int i = 0; i <= 1; ++i) { // Flag SingleServer 
        for (int j = 0; j <= 1; ++j) { // Flag Asynchronous
            for (int k = 0; k <=1; ++k) { // Flag ShortConnection
                TestSuccessSelective(i, j, k);
            }
        }
    }
}

TEST_F(ChannelTest, skip_parallel) {
    for (int i = 0; i <= 1; ++i) { // Flag SingleServer 
        for (int j = 0; j <= 1; ++j) { // Flag Asynchronous
            for (int k = 0; k <=1; ++k) { // Flag ShortConnection
                TestSkipParallel(i, j, k);
            }
        }
    }
}

TEST_F(ChannelTest, success_parallel2) {
    for (int i = 0; i <= 1; ++i) { // Flag SingleServer 
        for (int j = 0; j <= 1; ++j) { // Flag Asynchronous
            for (int k = 0; k <=1; ++k) { // Flag ShortConnection
                TestSuccessParallel2(i, j, k);
            }
        }
    }
}

TEST_F(ChannelTest, success_limit_parallel) {
    for (int i = 0; i <= 1; ++i) { // Flag SingleServer
        for (int j = 0; j <= 1; ++j) { // Flag Asynchronous
            for (int k = 0; k <=1; ++k) { // Flag ShortConnection
                TestSuccessLimitParallel(i, j, k);
            }
        }
    }
}

TEST_F(ChannelTest, cancel_before_callmethod) {
    for (int i = 0; i <= 1; ++i) { // Flag SingleServer 
        for (int j = 0; j <= 1; ++j) { // Flag Asynchronous
            for (int k = 0; k <=1; ++k) { // Flag ShortConnection
                CancelBeforeCallMethod(i, j, k);
            }
        }
    }
}

TEST_F(ChannelTest, cancel_before_callmethod_parallel) {
    for (int i = 0; i <= 1; ++i) { // Flag SingleServer 
        for (int j = 0; j <= 1; ++j) { // Flag Asynchronous
            for (int k = 0; k <=1; ++k) { // Flag ShortConnection
                CancelBeforeCallMethodParallel(i, j, k);
            }
        }
    }
}

TEST_F(ChannelTest, cancel_before_callmethod_selective) {
    for (int i = 0; i <= 1; ++i) { // Flag SingleServer 
        for (int j = 0; j <= 1; ++j) { // Flag Asynchronous
            for (int k = 0; k <=1; ++k) { // Flag ShortConnection
                CancelBeforeCallMethodSelective(i, j, k);
            }
        }
    }
}

TEST_F(ChannelTest, cancel_during_callmethod) {
    for (int i = 0; i <= 1; ++i) { // Flag SingleServer 
        for (int j = 0; j <= 1; ++j) { // Flag Asynchronous
            for (int k = 0; k <=1; ++k) { // Flag ShortConnection
                CancelDuringCallMethod(i, j, k);
            }
        }
    }
}

TEST_F(ChannelTest, cancel_during_callmethod_parallel) {
    for (int i = 0; i <= 1; ++i) { // Flag SingleServer 
        for (int j = 0; j <= 1; ++j) { // Flag Asynchronous
            for (int k = 0; k <=1; ++k) { // Flag ShortConnection
                CancelDuringCallMethodParallel(i, j, k);
            }
        }
    }
}

TEST_F(ChannelTest, cancel_during_callmethod_selective) {
    for (int i = 0; i <= 1; ++i) { // Flag SingleServer 
        for (int j = 0; j <= 1; ++j) { // Flag Asynchronous
            for (int k = 0; k <=1; ++k) { // Flag ShortConnection
                CancelDuringCallMethodSelective(i, j, k);
            }
        }
    }
}

TEST_F(ChannelTest, cancel_after_callmethod) {
    for (int i = 0; i <= 1; ++i) { // Flag SingleServer 
        for (int j = 0; j <= 1; ++j) { // Flag Asynchronous
            for (int k = 0; k <=1; ++k) { // Flag ShortConnection
                CancelAfterCallMethod(i, j, k);
            }
        }
    }
}

TEST_F(ChannelTest, cancel_after_callmethod_parallel) {
    for (int i = 0; i <= 1; ++i) { // Flag SingleServer 
        for (int j = 0; j <= 1; ++j) { // Flag Asynchronous
            for (int k = 0; k <=1; ++k) { // Flag ShortConnection
                CancelAfterCallMethodParallel(i, j, k);
            }
        }
    }
}

TEST_F(ChannelTest, request_not_init) {
    for (int i = 0; i <= 1; ++i) { // Flag SingleServer 
        for (int j = 0; j <= 1; ++j) { // Flag Asynchronous
            for (int k = 0; k <=1; ++k) { // Flag ShortConnection
                TestRequestNotInit(i, j, k);
            }
        }
    }
}

TEST_F(ChannelTest, request_not_init_parallel) {
    for (int i = 0; i <= 1; ++i) { // Flag SingleServer 
        for (int j = 0; j <= 1; ++j) { // Flag Asynchronous
            for (int k = 0; k <=1; ++k) { // Flag ShortConnection
                TestRequestNotInitParallel(i, j, k);
            }
        }
    }
}

TEST_F(ChannelTest, request_not_init_selective) {
    for (int i = 0; i <= 1; ++i) { // Flag SingleServer 
        for (int j = 0; j <= 1; ++j) { // Flag Asynchronous
            for (int k = 0; k <=1; ++k) { // Flag ShortConnection
                TestRequestNotInitSelective(i, j, k);
            }
        }
    }
}

TEST_F(ChannelTest, timeout) {
    for (int i = 0; i <= 1; ++i) { // Flag SingleServer 
        for (int j = 0; j <= 1; ++j) { // Flag Asynchronous
            for (int k = 0; k <=1; ++k) { // Flag ShortConnection
                TestRPCTimeout(i, j, k);
            }
        }
    }
}

TEST_F(ChannelTest, timeout_parallel) {
    for (int i = 0; i <= 1; ++i) { // Flag SingleServer 
        for (int j = 0; j <= 1; ++j) { // Flag Asynchronous
            for (int k = 0; k <=1; ++k) { // Flag ShortConnection
                TestRPCTimeoutParallel(i, j, k);
            }
        }
    }
}

TEST_F(ChannelTest, timeout_still_checks_sub_channels_parallel) {
    for (int i = 0; i <= 1; ++i) { // Flag SingleServer 
        for (int j = 0; j <= 1; ++j) { // Flag Asynchronous
            for (int k = 0; k <=1; ++k) { // Flag ShortConnection
                TimeoutStillChecksSubChannelsParallel(i, j, k);
            }
        }
    }
}

TEST_F(ChannelTest, timeout_selective) {
    for (int i = 0; i <= 1; ++i) { // Flag SingleServer 
        for (int j = 0; j <= 1; ++j) { // Flag Asynchronous
            for (int k = 0; k <=1; ++k) { // Flag ShortConnection
                TestRPCTimeoutSelective(i, j, k);
            }
        }
    }
}

TEST_F(ChannelTest, backuprequest_selective) {
    for (int i = 0; i <= 1; ++i) { // Flag SingleServer 
        for (int j = 0; j <= 1; ++j) { // Flag Asynchronous
            for (int k = 0; k <=1; ++k) { // Flag ShortConnection
                TestBackupRequestSelective(i, j, k);
            }
        }
    }
}

TEST_F(ChannelTest, backuprequest_selective_response_race) {
    TestBackupRequestSelectiveResponseRace();
}

TEST_F(ChannelTest, close_fd) {
    for (int i = 0; i <= 1; ++i) { // Flag SingleServer 
        for (int j = 0; j <= 1; ++j) { // Flag Asynchronous
            for (int k = 0; k <=1; ++k) { // Flag ShortConnection
                TestCloseFD(i, j, k);
            }
        }
    }
}

TEST_F(ChannelTest, close_fd_parallel) {
    for (int i = 0; i <= 1; ++i) { // Flag SingleServer 
        for (int j = 0; j <= 1; ++j) { // Flag Asynchronous
            for (int k = 0; k <=1; ++k) { // Flag ShortConnection
                TestCloseFDParallel(i, j, k);
            }
        }
    }
}

TEST_F(ChannelTest, close_fd_selective) {
    for (int i = 0; i <= 1; ++i) { // Flag SingleServer 
        for (int j = 0; j <= 1; ++j) { // Flag Asynchronous
            for (int k = 0; k <=1; ++k) { // Flag ShortConnection
                TestCloseFDSelective(i, j, k);
            }
        }
    }
}

TEST_F(ChannelTest, server_fail) {
    for (int i = 0; i <= 1; ++i) { // Flag SingleServer 
        for (int j = 0; j <= 1; ++j) { // Flag Asynchronous
            for (int k = 0; k <=1; ++k) { // Flag ShortConnection
                TestServerFail(i, j, k);
            }
        }
    }
}

TEST_F(ChannelTest, server_fail_parallel) {
    for (int i = 0; i <= 1; ++i) { // Flag SingleServer 
        for (int j = 0; j <= 1; ++j) { // Flag Asynchronous
            for (int k = 0; k <=1; ++k) { // Flag ShortConnection
                TestServerFailParallel(i, j, k);
            }
        }
    }
}

TEST_F(ChannelTest, server_fail_selective) {
    for (int i = 0; i <= 1; ++i) { // Flag SingleServer 
        for (int j = 0; j <= 1; ++j) { // Flag Asynchronous
            for (int k = 0; k <=1; ++k) { // Flag ShortConnection
                TestServerFailSelective(i, j, k);
            }
        }
    }
}

TEST_F(ChannelTest, authentication) {
    for (int i = 0; i <= 1; ++i) { // Flag SingleServer 
        for (int j = 0; j <= 1; ++j) { // Flag Asynchronous
            for (int k = 0; k <=1; ++k) { // Flag ShortConnection
                TestAuthentication(i, j, k);
            }
        }
    }
}

TEST_F(ChannelTest, authentication_parallel) {
    for (int i = 0; i <= 1; ++i) { // Flag SingleServer 
        for (int j = 0; j <= 1; ++j) { // Flag Asynchronous
            for (int k = 0; k <=1; ++k) { // Flag ShortConnection
                TestAuthenticationParallel(i, j, k);
            }
        }
    }
}

TEST_F(ChannelTest, authentication_selective) {
    for (int i = 0; i <= 1; ++i) { // Flag SingleServer 
        for (int j = 0; j <= 1; ++j) { // Flag Asynchronous
            for (int k = 0; k <=1; ++k) { // Flag ShortConnection
                TestAuthenticationSelective(i, j, k);
            }
        }
    }
}

TEST_F(ChannelTest, retry) {
    for (int i = 0; i <= 1; ++i) { // Flag SingleServer 
        for (int j = 0; j <= 1; ++j) { // Flag Asynchronous
            for (int k = 0; k <=1; ++k) { // Flag ShortConnection
                TestRetry(i, j, k);
            }
        }
    }
}

TEST_F(ChannelTest, retry_other_servers) {
    for (int j = 0; j <= 1; ++j) { // Flag Asynchronous
        for (int k = 0; k <=1; ++k) { // Flag ShortConnection
            TestRetryOtherServer(j, k);
        }
    }
}

TEST_F(ChannelTest, retry_backoff) {
    for (int j = 0; j <= 1; ++j) { // Flag Asynchronous
        for (int k = 0; k <= 1; ++k) { // Flag ShortConnection
            for (int l = 0; l <= 1; ++l) { // Flag FixedRetryBackoffPolicy or JitteredRetryBackoffPolicy
                for (int m = 0; m <= 1; ++m) { // Flag retry backoff in bthread or pthread
                    if (m % 2 == 0) {
                        bthread_t th;
                        bthread_attr_t attr = BTHREAD_ATTR_NORMAL;
                        std::unique_ptr<TestRetryBackoffInfo> test_retry_backoff(
                                new TestRetryBackoffInfo(this, j, k, l));
                        // Retry backoff in bthread.
                        bthread_start_background(&th, &attr, TestRetryBackoffBthread, test_retry_backoff.get());
                        bthread_join(th, NULL);
                    } else {
                        // Retry backoff in pthread.
                        TestRetryBackoff(j, k, l, true);
                    }
                }
            }
        }
    }
}

TEST_F(ChannelTest, backup_request) {
    for (int i = 0; i <= 1; ++i) { // Flag SingleServer
        for (int j = 0; j <= 1; ++j) { // Flag Asynchronous
            for (int k = 0; k <= 1; ++k) { // Flag ShortConnection
                TestBackupRequest(i, j, k);
            }
        }
    }
}

TEST_F(ChannelTest, backup_request_policy) {
    for (int i = 0; i <= 1; ++i) { // Flag SingleServer
        for (int j = 0; j <= 1; ++j) { // Flag Asynchronous
            for (int k = 0; k <= 1; ++k) { // Flag ShortConnection
                TestBackupRequestPolicy(i, j, k);
            }
        }
    }
}

TEST_F(ChannelTest, selective_channel_ignores_late_subdone_after_timeout) {
    DelayedCloseEchoService service;
    brpc::Server server;
    ASSERT_EQ(0, server.AddService(&service, brpc::SERVER_DOESNT_OWN_SERVICE));
    ASSERT_EQ(0, server.Start("127.0.0.1:0", NULL));

    brpc::SelectiveChannel channel;
    ASSERT_EQ(0, channel.Init("rr", NULL));

    brpc::ChannelOptions options;
    options.timeout_ms = 100;
    for (int i = 0; i < 2; ++i) {
        brpc::Channel* sub_channel = new brpc::Channel;
        ASSERT_EQ(0, sub_channel->Init(server.listen_address(), &options));
        ASSERT_EQ(0, channel.AddChannel(sub_channel, NULL));
    }

    brpc::Controller cntl;
    cntl.set_max_retry(3);
    cntl.set_backup_request_ms(1);
    cntl.set_timeout_ms(10);

    test::EchoRequest req;
    test::EchoResponse res;
    req.set_message(__FUNCTION__);
    req.set_sleep_us(50000);
    CallMethod(&channel, &cntl, &req, &res, true);

    ASSERT_EQ(brpc::ERPCTIMEDOUT, cntl.ErrorCode()) << cntl.ErrorText();
    ASSERT_TRUE(cntl.has_backup_request());
    ASSERT_GE(cntl.retried_count(), 1);

    // Let the delayed sub-calls close their connections after the main RPC
    // has already timed out. This used to re-enter retry/backup from
    // SubDone::Run() on a partially torn-down controller.
    bthread_usleep(120000);

    EXPECT_EQ(brpc::ERPCTIMEDOUT, cntl.ErrorCode()) << cntl.ErrorText();
    server.Stop(0);
    server.Join();
}

TEST_F(ChannelTest, multiple_threads_single_channel) {
    srand(time(NULL));
    ASSERT_EQ(0, StartAccept(_ep));
    MyAuthenticator auth;
    const int NUM = 10;
    const int COUNT = 10000;
    pthread_t tids[NUM];

    // Cause massive connect/close log if setting to true
    bool short_connection = false;
    for (int single_server = 0; single_server <= 1; ++single_server) {
        for (int need_auth = 0; need_auth <= 1; ++need_auth) {
            for (int async = 0; async <= 1; ++async) {
                std::cout << " *** short=" << short_connection
                          << " single=" << single_server
                          << " auth=" << need_auth
                          << " async=" << async << std::endl;
                brpc::Channel channel;
                SetUpChannel(&channel, single_server, 
                             short_connection, (need_auth ? &auth : NULL));
                for (int i = 0; i < NUM; ++i) {
                    google::protobuf::Closure* thrd_func = 
                        brpc::NewCallback(
                            this, &ChannelTest::RPCThread, 
                            (brpc::ChannelBase*)&channel,
                            (bool)async, COUNT);
                    EXPECT_EQ(0, pthread_create(&tids[i], NULL,
                                                RunClosure, thrd_func));
                }
                for (int i = 0; i < NUM; ++i) {
                    pthread_join(tids[i], NULL);
                }
            }
        }
    }
}

TEST_F(ChannelTest, multiple_threads_multiple_channels) {
    srand(time(NULL));
    ASSERT_EQ(0, StartAccept(_ep));
    MyAuthenticator auth;
    const int NUM = 10;
    const int COUNT = 10000;
    pthread_t tids[NUM];

    // Cause massive connect/close log if setting to true
    bool short_connection = false;

    for (int single_server = 0; single_server <= 1; ++single_server) {
        for (int need_auth = 0; need_auth <= 1; ++need_auth) {
            for (int async = 0; async <= 1; ++async) {
                std::cout << " *** short=" << short_connection
                          << " single=" << single_server
                          << " auth=" << need_auth
                          << " async=" << async << std::endl;
                for (int i = 0; i < NUM; ++i) {
                    google::protobuf::Closure* thrd_func = 
                        brpc::NewCallback<
                        ChannelTest, ChannelTest*,
                        bool, bool, bool, const brpc::Authenticator*, int>
                        (this, &ChannelTest::RPCThread, single_server,
                         async, short_connection, (need_auth ? &auth : NULL), COUNT);
                    EXPECT_EQ(0, pthread_create(&tids[i], NULL,
                                                RunClosure, thrd_func));
                }
                for (int i = 0; i < NUM; ++i) {
                    pthread_join(tids[i], NULL);
                }
            }
        }
    }
}

TEST_F(ChannelTest, clear_attachment_after_retry) {
    for (int j = 0; j <= 1; ++j) {
        for (int k = 0; k <= 1; ++k) {
            TestAttachment(j, k);
        }
    }
}

TEST_F(ChannelTest, destroy_channel) {
    for (int i = 0; i <= 1; ++i) {
        for (int j = 0; j <= 1; ++j) {
            TestDestroyChannel(i, j);
        }
    }
}

TEST_F(ChannelTest, destroy_channel_parallel) {
    for (int i = 0; i <= 1; ++i) {
        for (int j = 0; j <= 1; ++j) {
            TestDestroyChannelParallel(i, j);
        }
    }
}

TEST_F(ChannelTest, destroy_channel_selective) {
    for (int i = 0; i <= 1; ++i) {
        for (int j = 0; j <= 1; ++j) {
            TestDestroyChannelSelective(i, j);
        }
    }
}

TEST_F(ChannelTest, sizeof) {
    LOG(INFO) << "Size of Channel is " << sizeof(brpc::Channel)
               << ", Size of ParallelChannel is " << sizeof(brpc::ParallelChannel)
               << ", Size of Controller is " << sizeof(brpc::Controller)
               << ", Size of vector is " << sizeof(std::vector<brpc::Controller>);
}

brpc::Channel g_chan;

TEST_F(ChannelTest, global_channel_should_quit_successfully) {
    g_chan.Init("bns://qa-pbrpc.SAT.tjyx", "rr", NULL);
}

TEST_F(ChannelTest, unused_call_id) {
    {
        brpc::Controller cntl;
    }
    {
        brpc::Controller cntl;
        cntl.Reset();
    }
    brpc::CallId cid1 = { 0 };
    {
        brpc::Controller cntl;
        cid1 = cntl.call_id();
    }
    ASSERT_EQ(EINVAL, bthread_id_error(cid1, ECANCELED));

    {
        brpc::CallId cid2 = { 0 };
        brpc::Controller cntl;
        cid2 = cntl.call_id();
        cntl.Reset();
        ASSERT_EQ(EINVAL, bthread_id_error(cid2, ECANCELED));
    }
}

TEST_F(ChannelTest, adaptive_connection_type) {
    brpc::AdaptiveConnectionType ctype;
    ASSERT_EQ(brpc::CONNECTION_TYPE_UNKNOWN, ctype);
    ASSERT_FALSE(ctype.has_error());
    ASSERT_STREQ("unknown", ctype.name());

    ctype = brpc::CONNECTION_TYPE_SINGLE;
    ASSERT_EQ(brpc::CONNECTION_TYPE_SINGLE, ctype);
    ASSERT_STREQ("single", ctype.name());

    ctype = "shorT";
    ASSERT_EQ(brpc::CONNECTION_TYPE_SHORT, ctype);
    ASSERT_STREQ("short", ctype.name());
    
    ctype = "PooLed";
    ASSERT_EQ(brpc::CONNECTION_TYPE_POOLED, ctype);
    ASSERT_STREQ("pooled", ctype.name());

    ctype = "SINGLE";
    ASSERT_EQ(brpc::CONNECTION_TYPE_SINGLE, ctype);
    ASSERT_FALSE(ctype.has_error());
    ASSERT_STREQ("single", ctype.name());

    ctype = "blah";
    ASSERT_EQ(brpc::CONNECTION_TYPE_UNKNOWN, ctype);
    ASSERT_TRUE(ctype.has_error());
    ASSERT_STREQ("unknown", ctype.name());

    ctype = "single";
    ASSERT_EQ(brpc::CONNECTION_TYPE_SINGLE, ctype);
    ASSERT_FALSE(ctype.has_error());
    ASSERT_STREQ("single", ctype.name());
}

TEST_F(ChannelTest, adaptive_protocol_type) {
    brpc::AdaptiveProtocolType ptype;
    ASSERT_EQ(brpc::PROTOCOL_UNKNOWN, ptype);
    ASSERT_STREQ("unknown", ptype.name());
    ASSERT_FALSE(ptype.has_param());
    ASSERT_EQ("", ptype.param());

    ptype = brpc::PROTOCOL_HTTP;
    ASSERT_EQ(brpc::PROTOCOL_HTTP, ptype);
    ASSERT_STREQ("http", ptype.name());
    ASSERT_FALSE(ptype.has_param());
    ASSERT_EQ("", ptype.param());

    ptype = "http:xyz ";
    ASSERT_EQ(brpc::PROTOCOL_HTTP, ptype);
    ASSERT_STREQ("http", ptype.name());
    ASSERT_TRUE(ptype.has_param());
    ASSERT_EQ("xyz ", ptype.param());

    ptype = "HuLu_pbRPC";
    ASSERT_EQ(brpc::PROTOCOL_HULU_PBRPC, ptype);
    ASSERT_STREQ("hulu_pbrpc", ptype.name());
    ASSERT_FALSE(ptype.has_param());
    ASSERT_EQ("", ptype.param());
    
    ptype = "blah";
    ASSERT_EQ(brpc::PROTOCOL_UNKNOWN, ptype);
    ASSERT_STREQ("blah", ptype.name());
    ASSERT_FALSE(ptype.has_param());
    ASSERT_EQ("", ptype.param());

    ptype = "Baidu_STD";
    ASSERT_EQ(brpc::PROTOCOL_BAIDU_STD, ptype);
    ASSERT_STREQ("baidu_std", ptype.name());
    ASSERT_FALSE(ptype.has_param());
    ASSERT_EQ("", ptype.param());
}

class RateLimitedBackupPolicyTest : public ::testing::Test {};

TEST_F(RateLimitedBackupPolicyTest, InvalidBackupRequestMs) {
    brpc::RateLimitedBackupPolicyOptions opts;
    opts.backup_request_ms = -2;
    ASSERT_EQ(NULL, brpc::CreateRateLimitedBackupPolicy(opts));
}

TEST_F(RateLimitedBackupPolicyTest, InvalidMaxBackupRatioZero) {
    brpc::RateLimitedBackupPolicyOptions opts;
    opts.backup_request_ms = 100;
    opts.max_backup_ratio = 0.0;
    ASSERT_EQ(NULL, brpc::CreateRateLimitedBackupPolicy(opts));
}

TEST_F(RateLimitedBackupPolicyTest, InvalidMaxBackupRatioNegative) {
    brpc::RateLimitedBackupPolicyOptions opts;
    opts.backup_request_ms = 100;
    opts.max_backup_ratio = -0.1;
    ASSERT_EQ(NULL, brpc::CreateRateLimitedBackupPolicy(opts));
}

TEST_F(RateLimitedBackupPolicyTest, InvalidMaxBackupRatioAboveOne) {
    brpc::RateLimitedBackupPolicyOptions opts;
    opts.backup_request_ms = 100;
    opts.max_backup_ratio = 1.001;
    ASSERT_EQ(NULL, brpc::CreateRateLimitedBackupPolicy(opts));
}

TEST_F(RateLimitedBackupPolicyTest, InvalidWindowSizeTooSmall) {
    brpc::RateLimitedBackupPolicyOptions opts;
    opts.backup_request_ms = 100;
    opts.window_size_seconds = 0;
    ASSERT_EQ(NULL, brpc::CreateRateLimitedBackupPolicy(opts));
}

TEST_F(RateLimitedBackupPolicyTest, InvalidWindowSizeTooLarge) {
    brpc::RateLimitedBackupPolicyOptions opts;
    opts.backup_request_ms = 100;
    opts.window_size_seconds = 3601;
    ASSERT_EQ(NULL, brpc::CreateRateLimitedBackupPolicy(opts));
}

TEST_F(RateLimitedBackupPolicyTest, InvalidUpdateIntervalTooSmall) {
    brpc::RateLimitedBackupPolicyOptions opts;
    opts.backup_request_ms = 100;
    opts.update_interval_seconds = 0;
    ASSERT_EQ(NULL, brpc::CreateRateLimitedBackupPolicy(opts));
}

TEST_F(RateLimitedBackupPolicyTest, ValidMinusOneBackupRequestMsInherits) {
    brpc::RateLimitedBackupPolicyOptions opts;
    opts.backup_request_ms = -1;
    std::unique_ptr<brpc::BackupRequestPolicy> p(
        brpc::CreateRateLimitedBackupPolicy(opts));
    ASSERT_TRUE(p != NULL);
    ASSERT_EQ(-1, p->GetBackupRequestMs(NULL));
}

TEST_F(RateLimitedBackupPolicyTest, ValidMaxRatioAtBoundary) {
    brpc::RateLimitedBackupPolicyOptions opts;
    opts.backup_request_ms = 50;
    opts.max_backup_ratio = 1.0;
    std::unique_ptr<brpc::BackupRequestPolicy> p(
        brpc::CreateRateLimitedBackupPolicy(opts));
    ASSERT_TRUE(p != NULL);
    // With max_backup_ratio=1.0 and true cold start (total==0, backup==0),
    // ShouldAllow() sets ratio=0.0 (free pass). The conservative ratio=1.0
    // path only applies when backup>0 but total==0 (latency spike with no
    // completions yet). At absolute cold start DoBackup() must return true.
    ASSERT_TRUE(p->DoBackup(NULL));   // cold start: ratio=0.0 < 1.0, allow
}

TEST_F(RateLimitedBackupPolicyTest, ColdStartAllowsBackup) {
    brpc::RateLimitedBackupPolicyOptions opts;
    opts.backup_request_ms = 10;
    opts.max_backup_ratio = 0.1;
    opts.update_interval_seconds = 1;
    std::unique_ptr<brpc::BackupRequestPolicy> p(
        brpc::CreateRateLimitedBackupPolicy(opts));
    ASSERT_TRUE(p != NULL);
    ASSERT_TRUE(p->DoBackup(NULL));
}

// After the first backup fires (backup_count=1, total_count=0), once the
// update interval elapses the ratio is refreshed via the conservative path
// (total==0 → ratio=1.0), which exceeds max_backup_ratio < 1.0, so
// subsequent DoBackup() calls are suppressed until an RPC leg completes.
TEST_F(RateLimitedBackupPolicyTest, AfterColdStartBackupSuppressedUntilRpcCompletes) {
    brpc::RateLimitedBackupPolicyOptions opts;
    opts.backup_request_ms = 10;
    opts.max_backup_ratio = 0.1;
    opts.window_size_seconds = 1;
    opts.update_interval_seconds = 1;
    std::unique_ptr<brpc::BackupRequestPolicy> p(
        brpc::CreateRateLimitedBackupPolicy(opts));
    ASSERT_TRUE(p != NULL);
    // First call fires (cold start: total=0, backup=0 → ratio=0.0 → allow).
    ASSERT_TRUE(p->DoBackup(NULL));
    // Wait for the update interval to elapse so the ratio refreshes.
    // After refresh: total=0 but backup=1 → conservative path sets ratio=1.0,
    // which is >= max_backup_ratio (0.1), so DoBackup() must return false.
    bthread_usleep(1200000); // 1.2s > update_interval_seconds=1
    ASSERT_FALSE(p->DoBackup(NULL));
}

// After the ratio rises above the threshold, calling OnRPCEnd() many times
// drives total_count up relative to backup_count. Once the ratio refreshes
// below max_backup_ratio, DoBackup() should allow backups again.
TEST_F(RateLimitedBackupPolicyTest, OnRPCEndDrivesRatioDownAndReAllows) {
    brpc::RateLimitedBackupPolicyOptions opts;
    opts.backup_request_ms = 10;
    opts.max_backup_ratio = 0.5;
    opts.window_size_seconds = 1;
    opts.update_interval_seconds = 1;
    std::unique_ptr<brpc::BackupRequestPolicy> p(
        brpc::CreateRateLimitedBackupPolicy(opts));
    ASSERT_TRUE(p != NULL);
    // Fire many backup decisions so backup_count >> total_count,
    // pushing the ratio above max_backup_ratio.
    for (int i = 0; i < 20; ++i) {
        p->DoBackup(NULL);
    }
    // Wait for update interval so the ratio is refreshed above threshold.
    bthread_usleep(1200000); // 1.2s
    ASSERT_FALSE(p->DoBackup(NULL));
    // Now complete many more RPCs than backups fired to bring ratio below 0.5.
    // 20 backup decisions already counted; need total_count > 20/0.5 = 40.
    for (int i = 0; i < 50; ++i) {
        p->OnRPCEnd(NULL);
    }
    // Wait for the ratio cache to refresh.
    bthread_usleep(1200000); // 1.2s
    // Ratio is now ~20/50 = 0.4 < max_backup_ratio (0.5), so backup is re-allowed.
    ASSERT_TRUE(p->DoBackup(NULL));
}

} //namespace
