/** * @brief DataStore 的具备协程能力的装饰器类型。 * @attention 除了定义变量语句和 Update() 以外，其他的操作都需要在协程中调用。 */class CoroutineDataStore : public Updateable{public:    CoroutineDataStore(DataStore* data_store, CoroutineSchedule* schedule);    virtual ~CoroutineDataStore();
    int Init(Config* cfg, std::string* err_msg);
    /**     * 读取一个数据对象，通过 key ，把数据放入到输出参数 value。     * 此函数会在调用过程中使用协程的 yield 出去。     */    int Get(const std::string&key, Serializable* value);
    /**     * 写入一个数据对象，写入 key ，value     * 写入结果从返回值获得，返回 0 表示成功，其他值表示失败。     * 此函数会在调用过程中使用协程的 yield 出去。     */    int Put(const std::string&key, const Serializable& value);
    /**     * 删除一个数据对象，通过 key     * 写入结果从返回值获得，返回 0 表示成功，其他值表示失败。     * 此函数会在调用过程中使用协程的 yield 出去     */    int Remove(const std::string& key);
    int Update();
private:    DataStore* data_store_;    CoroutineSchedule* schedule_;};

/** * 代表一个应用程序组件，所有的应用程序组件应该继承此对象 */class Component : public Updateable {
public:    Component();    virtual ~Component();
    /**     * 返回此组件的名字     * @return 名字     */    virtual std::string GetName() = 0;
    /**     * 初始化过程会调用此方法     * @param app 进程对象     * @param cfg 配置     * @return 返回 0 表示成功，其他表示失败     */    virtual int Init(Application* app, Config* cfg);
    /**     * 更新过程会调用此方法     * @return 返回更新处理的负载，0 表示没有负载，负数表示出错或需要停止对此组件进行更新     */    virtual int Update();
    /**     * 应用停止时会调用此方法     * @return 返回 0 表示可以退出，返回 >0 表示需要等待返回值的秒数，返回 < 0 表示出错退出     */    virtual int Stop() {        return 0;    };
    /**     * 设置组件被加入的应用程序，用于让组件继承者能简单的获取 Application 对象     * @note 如果一个组件被加入多个不同的 Application，必须使用 @see Init() 方法来具体保存 Application 对象，     * 因为此处修改的成员对象 app_ 将是最后一次被添加进的 Application 对象。     * @param app 要设置的 Application 对象。     */    void set_app(Application* app){        app_ = app;    }
protected:    Application* app_;};

class LocalServerApp : public Application { public:  LocalServerApp();  virtual ~LocalServerApp();  virtual int Init(Config* cfg = NULL);  virtual int Exit();  void set_transport(Transport* transport);  void set_protocol(Protocol* protocol);  void set_processor(Processor* processor);
 private:  Transport* transport_;  Protocol* protocol_;  Processor* processor_;  Server* server_;};

tcp://1.1.1.1:8888/tlv

/** * @brief 集群中心客户端 * 每个 DenOS 进程启动时，都会向 ZooKeeper 注册自己的服务。 * 此类型的对象，就是作为每个进程中，代表集群中心的存在。 * ZooKeeper 默认根据 2 个 tick （心跳），大概为 3 秒，是否收到，来决定客户端是否死掉。 * 在 DenOS 中，可以使用配置项目 ZK_RECV_TIMEOUT 参数（单位为毫秒）来决定这个超时时间。 */class Center : public ZooKeeperMap {friend    void ContractsWatcher(zhandle_t *zh, int type, int state, const char *path,                          void *watcherCtx);public:    int last_error_code_;  // 用来测试最后一个操作是否成功的变量    std::map<std::string, std::string> service_process_;  // key: 服务名字，value  服务节点名字
    /**     * @brief 构造一个集群中心客户端     * @param urls ZooKeeper 的连接参数，形如："127.0.0.1:2181,10.1.2.3:2182,192.168.3.23:2183"     */    Center(const std::string& urls = "127.0.0.1:2181");    virtual ~Center();
    /// 驱动整个异步流程    virtual int Update();
    /**     * 往集群中声明注册服务     * @param name 服务的名字     * @param contract 服务的通信方式     * @return 返回 0 表示已经发起注册流程，其他值表示失败     */    int RegisterService(const std::string& name, const Contract& contract);
    /**     * @brief 注册一个类作为对应协议字符串名字     * 如果这个类是 Connector 的子类，必须要有一个形如 XXConnector(const string& p1, int p2) 这样的构造器。     * 或者这个类是 Protocol 的子类，必须要有一个无参数构造器。     * @param reg_name 协议字符串名字，如 tcp/udp/kcp/tconnd 或者 tlv/line/tdr     */    template<typename T>    void RegProto(const std::string& reg_name) {        if (reg_name.empty())            return;        constructors_[reg_name] = new DefaultConstructor<T>();    }
    template<typename T>    void RegConn(const std::string& reg_name) {        if (reg_name.empty())            return;        constructors_[reg_name] = new StrIntConstructor<T>();    }
    /**     * @brief 查询一个服务去发起请求     * 注意这是一个异步的接口，有可能会返回 -1 表示服务合同还未拿到。需要重复的去获取。     * @param name 服务的名字     * @param callback 当获得对应的服务的客户端的回调     * @param client_cb 预期每个新的 Client 所注册的默认回调，用来接收连接、中断、收听通知。     * @param route_param 用来传给路由器的自定义路由相关数据     * @return  如果返回 0 表示成功，失败则会是其他数值     */    int QueryService(const std::string& name,                     GetServiceClientCallback* callback,                     ClientCallback* client_cb = NULL, Router* router = NULL,                     void* route_param = NULL);
    /**     * 根据合同缓存获得客户端对象     * @param cache 合同缓存对象     * @param client_cb 预期每个新的 Client 所注册的默认回调，用来接收连接、中断、收听通知。     * @param router 路由器对象     * @param route_param 路由参数     * @return 客户端对象指针，无需主动 delete，因为会缓存起来     */    Client* GetClientByContracts(ContractCache* cache,                                 ClientCallback* client_cb, Router* router,                                 void* route_param);
    /**     * 获得存放集群的 ZK 基础路径     * @return ZNode 基础路径     */    inline const std::string& cluster_prefix() const {        return CLUSTER_PREFIX;    }
    /**     * 获得建立进程用的 ZK 标记 (Create flags)     * @return zk 的 create flags     */    inline int process_flags() const {        return process_flags_;    }
    /// 在 ZK 写入本进程对此服务的合约    void AddProcessContract(const std::string& service_name,                            const std::string& contract_data);
    /// 清理相关对象    void CloseClient(Client* client);
    void SetContractsCache(const std::string& service_name,                           ContractCache* cache);
    /*---------------------------------- 继承自 ZooKeeper 为了实现功能用 -----------------------------------*/    /// 建立存储节点父目录时，增加一个监听器，监听这些节点增加和删除变化    virtual void CreatePrefixNode();
    /// 初始化 zookeeper 客户端连接，会修改 ZKMAP_KEY_PREFIX 为集群专用路径    virtual int Init(Config* config = NULL);
    /**     * 清理掉生成的客户端对象     * @param client 客户端对象     * @param content 相关的服务名字     */    void ClearClientMember(Client* client, const std::string& content);
};

/** * @brief 路由器基类 */class Router {public:    Router();    virtual ~Router();
    /**     * @brief 决定服务路由的接口     * 此接口默认实现是取 cache 中的第一个非空结果     * @param cache 需要选择的所有合同的缓存集合     * @param content 输出参数，具体选择的合同的内容     * @param route_param 用来提供给路由算法运行的额外参数     */    virtual void RouteToSevice(const ContractCache& cache, std::string* content,                               void* route_param = NULL);};

    // 获取客户端    DEBUG_LOG("========== Getting Client ==========");    Center* center = obj_pro_reg.center();    TestGetClientCallback callback;    TestClientCallback cli_cb;    center->QueryService(handler.service_name_, &callback, &cli_cb);    for (int i = 0; i < 100; i++) {        usleep(100);        svr->Update();    }    EXPECT_EQ(0, callback.err_code_);    EXPECT_TRUE(cli_cb.is_connected_);    ASSERT_TRUE(callback.client_ != NULL);
    // 访问服务器    DEBUG_LOG("========== Requesting Service ==========");    Client* client = callback.client_;    Request req;    string data("I love JMX!");    req.service = handler.service_name_;    req.SetData(data.c_str(), data.length());    client->SendRequest(&req, &cli_cb);    for (int i = 0; i < 100; i++) {        usleep(100);        svr->Update();    }    EXPECT_EQ(data, cli_cb.resp_data_);