高可靠对称节点（双星模式）

// 创建双星模式实例，使用本地（绑定）和远程（连接）端点来设置节点对。
bstar_t *bstar_new (int primary, char *local, char *remote);

// 销毁实例
void bstar_destroy (bstar_t **self_p);

// 返回底层的zloop反应堆，用以添加定时器、读取器、注册和取消等功能。
zloop_t *bstar_zloop (bstar_t *self);

// 注册投票读取器
int bstar_voter (bstar_t *self, char *endpoint, int type,
zloop_fn handler, void *arg);

// 注册状态机处理器
void bstar_new_master (bstar_t *self, zloop_fn handler, void *arg);
void bstar_new_slave (bstar_t *self, zloop_fn handler, void *arg);

// 开启反应堆，当回调函数返回-1，或进程收到SIGINT、SIGTERM信号时中止。
int bstar_start (bstar_t *self);

/*  =====================================================================
    bstar - Binary Star reactor

    ---------------------------------------------------------------------
    Copyright (c) 1991-2011 iMatix Corporation <www.imatix.com>
    Copyright other contributors as noted in the AUTHORS file.

    This file is part of the ZeroMQ Guide: http://zguide.zeromq.org

    This is free software; you can redistribute it and/or modify it under
    the terms of the GNU Lesser General Public License as published by
    the Free Software Foundation; either version 3 of the License, or (at
    your option) any later version.

    This software is distributed in the hope that it will be useful, but
    WITHOUT ANY WARRANTY; without even the implied warranty of
    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
    Lesser General Public License for more details.

    You should have received a copy of the GNU Lesser General Public
    License along with this program. If not, see
    <http://www.gnu.org/licenses/>.
    =====================================================================
*/

#include "bstar.h"

//  服务器状态枚举
typedef enum {
    STATE_PRIMARY = 1,          //  主机，等待同伴连接
    STATE_BACKUP = 2,           //  备机，等待同伴连接
    STATE_ACTIVE = 3,           //  激活态，处理应用程序请求
    STATE_PASSIVE = 4           //  被动态，不接收请求
} state_t;

//  对话节点事件
typedef enum {
    PEER_PRIMARY = 1,           //  主机
    PEER_BACKUP = 2,            //  备机
    PEER_ACTIVE = 3,            //  激活态
    PEER_PASSIVE = 4,           //  被动态
    CLIENT_REQUEST = 5          //  客户端请求
} event_t;


//  发送状态信息的间隔时间
//  如果对方在两次心跳过后都没有应答，则视为断开
#define BSTAR_HEARTBEAT     1000        //  In msecs

//  类结构

struct _bstar_t {
    zctx_t *ctx;                //  私有上下文
    zloop_t *loop;              //  反应堆循环
    void *statepub;             //  状态发布者
    void *statesub;             //  状态订阅者
    state_t state;              //  当前状态
    event_t event;              //  当前事件
    int64_t peer_expiry;        //  判定节点死亡的时限
    zloop_fn *voter_fn;         //  投票套接字处理器
    void *voter_arg;            //  投票处理程序的参数
    zloop_fn *master_fn;        //  成为master时回调
    void *master_arg;           //  参数
    zloop_fn *slave_fn;         //  成为slave时回调
    void *slave_arg;            //  参数
};


//  ---------------------------------------------------------------------
//  执行有限状态机（将事件绑定至状态）；
//  发生异常时返回-1，正确时返回0。

static int
s_execute_fsm (bstar_t *self)
{
    int rc = 0;
    //  主机等待同伴连接
    //  该状态下接收CLIENT_REQUEST事件
    if (self->state == STATE_PRIMARY) {
        if (self->event == PEER_BACKUP) {
            zclock_log ("I: 已连接至备机（slave），可以作为master运行。");
            self->state = STATE_ACTIVE;
            if (self->master_fn)
                (self->master_fn) (self->loop, NULL, self->master_arg);
        }
        else
        if (self->event == PEER_ACTIVE) {
            zclock_log ("I: 已连接至备机（master），可以作为slave运行。");
            self->state = STATE_PASSIVE;
            if (self->slave_fn)
                (self->slave_fn) (self->loop, NULL, self->slave_arg);
        }
        else
        if (self->event == CLIENT_REQUEST) {
            zclock_log ("I: 收到客户端请求，可作为master运行。");
            self->state = STATE_ACTIVE;
            if (self->master_fn)
                (self->master_fn) (self->loop, NULL, self->master_arg);
        }
    }
    else
    //  备机等待同伴连接
    //  该状态下拒绝CLIENT_REQUEST事件
    if (self->state == STATE_BACKUP) {
        if (self->event == PEER_ACTIVE) {
            zclock_log ("I: 已连接至主机（master），可以作为slave运行。");
            self->state = STATE_PASSIVE;
            if (self->slave_fn)
                (self->slave_fn) (self->loop, NULL, self->slave_arg);
        }
        else
        if (self->event == CLIENT_REQUEST)
            rc = -1;
    }
    else
    //  服务器处于激活态
    //  该状态下接受CLIENT_REQUEST事件
    //  只有服务器死亡才会离开激活态
    if (self->state == STATE_ACTIVE) {
        if (self->event == PEER_ACTIVE) {
            //  若出现两台master，则抛出异常
            zclock_log ("E: 严重错误：双master。正在退出。");
            rc = -1;
        }
    }
    else
    //  服务器处于被动态
    //  若同伴已死，CLIENT_REQUEST事件将触发故障恢复
    if (self->state == STATE_PASSIVE) {
        if (self->event == PEER_PRIMARY) {
            //  同伴正在重启 - 转为激活态，同伴将转为被动态。
            zclock_log ("I: 主机（slave）正在重启，可作为master运行。");
            self->state = STATE_ACTIVE;
        }
        else
        if (self->event == PEER_BACKUP) {
            //  同伴正在重启 - 转为激活态，同伴将转为被动态。
            zclock_log ("I: 备机（slave）正在重启，可作为master运行。");
            self->state = STATE_ACTIVE;
        }
        else
        if (self->event == PEER_PASSIVE) {
            //  若出现两台slave，集群将无响应
            zclock_log ("E: 严重错误：双slave。正在退出");
            rc = -1;
        }
        else
        if (self->event == CLIENT_REQUEST) {
            //  若心跳超时，同伴将成为master；
            //  此行为由客户端请求触发。
            assert (self->peer_expiry > 0);
            if (zclock_time () >= self->peer_expiry) {
                //  同伴已死，转为激活态。
                zclock_log ("I: 故障恢复，可作为master运行。");
                self->state = STATE_ACTIVE;
            }
            else
                //  同伴还在，拒绝请求。
                rc = -1;
        }
        //  触发状态更改事件处理函数
        if (self->state == STATE_ACTIVE && self->master_fn)
            (self->master_fn) (self->loop, NULL, self->master_arg);
    }
    return rc;
}


//  ---------------------------------------------------------------------
//  反应堆事件处理程序

//  发送状态信息
int s_send_state (zloop_t *loop, void *socket, void *arg)
{
    bstar_t *self = (bstar_t *) arg;
    zstr_sendf (self->statepub, "%d", self->state);
    return 0;
}

//  接收状态信息，启动有限状态机
int s_recv_state (zloop_t *loop, void *socket, void *arg)
{
    bstar_t *self = (bstar_t *) arg;
    char *state = zstr_recv (socket);
    if (state) {
        self->event = atoi (state);
        self->peer_expiry = zclock_time () + 2 * BSTAR_HEARTBEAT;
        free (state);
    }
    return s_execute_fsm (self);
}

//  收到应用程序请求，判断是否接收
int s_voter_ready (zloop_t *loop, void *socket, void *arg)
{
    bstar_t *self = (bstar_t *) arg;
    //  如果能够处理请求，则调用函数
    self->event = CLIENT_REQUEST;
    if (s_execute_fsm (self) == 0) {
        puts ("CLIENT REQUEST");
        (self->voter_fn) (self->loop, socket, self->voter_arg);
    }
    else {
        //  销毁等待中的消息
        zmsg_t *msg = zmsg_recv (socket);
        zmsg_destroy (&msg);
    }
    return 0;
}


//  ---------------------------------------------------------------------
//  构造函数

bstar_t *
bstar_new (int primary, char *local, char *remote)
{
    bstar_t
        *self;

    self = (bstar_t *) zmalloc (sizeof (bstar_t));

    //  初始化双星
    self->ctx = zctx_new ();
    self->loop = zloop_new ();
    self->state = primary? STATE_PRIMARY: STATE_BACKUP;

    //  创建状态PUB套接字
    self->statepub = zsocket_new (self->ctx, ZMQ_PUB);
    zsocket_bind (self->statepub, local);

    //  创建状态SUB套接字
    self->statesub = zsocket_new (self->ctx, ZMQ_SUB);
    zsocket_connect (self->statesub, remote);

    //  设置基本的反应堆事件处理器
    zloop_timer (self->loop, BSTAR_HEARTBEAT, 0, s_send_state, self);
    zloop_reader (self->loop, self->statesub, s_recv_state, self);
    return self;
}


//  ---------------------------------------------------------------------
//  析构函数

void
bstar_destroy (bstar_t **self_p)
{
    assert (self_p);
    if (*self_p) {
        bstar_t *self = *self_p;
        zloop_destroy (&self->loop);
        zctx_destroy (&self->ctx);
        free (self);
        *self_p = NULL;
    }
}


//  ---------------------------------------------------------------------
//  返回底层zloop对象，用以添加额外的定时器、阅读器等。

zloop_t *
bstar_zloop (bstar_t *self)
{
    return self->loop;
}


//  ---------------------------------------------------------------------
//  创建套接字，连接至本地端点，注册成为阅读器；
//  只有当有限状态机允许时才会读取该套接字；
//  从该套接字获得的消息将作为一次“投票”；
//  我们要求双星模式中只有一个“投票”套接字。

int
bstar_voter (bstar_t *self, char *endpoint, int type, zloop_fn handler,
             void *arg)
{
    //  保存原始的回调函数和参数，稍后使用
    void *socket = zsocket_new (self->ctx, type);
    zsocket_bind (socket, endpoint);
    assert (!self->voter_fn);
    self->voter_fn = handler;
    self->voter_arg = arg;
    return zloop_reader (self->loop, socket, s_voter_ready, self);
}

//  ---------------------------------------------------------------------
//  注册状态变化事件处理器

void
bstar_new_master (bstar_t *self, zloop_fn handler, void *arg)
{
    assert (!self->master_fn);
    self->master_fn = handler;
    self->master_arg = arg;
}

void
bstar_new_slave (bstar_t *self, zloop_fn handler, void *arg)
{
    assert (!self->slave_fn);
    self->slave_fn = handler;
    self->slave_arg = arg;
}


//  ---------------------------------------------------------------------
//  启用或禁止跟踪信息
void bstar_set_verbose (bstar_t *self, Bool verbose)
{
    zloop_set_verbose (self->loop, verbose);
}


//  ---------------------------------------------------------------------
//  开启反应堆，当回调函数返回-1，或进程收到SIGINT、SIGTERM信号时中止。

int
bstar_start (bstar_t *self)
{
    assert (self->voter_fn);
    return zloop_start (self->loop);
}

//
//  双星模式服务端，使用bstar反应堆
//

//  直接编译，不建类库
#include "bstar.c"

//  Echo service
int s_echo (zloop_t *loop, void *socket, void *arg)
{
    zmsg_t *msg = zmsg_recv (socket);
    zmsg_send (&msg, socket);
    return 0;
}

int main (int argc, char *argv [])
{
    //  命令行参数可以为：
    //      -p  作为主机启动, at tcp://localhost:5001
    //      -b  作为备机启动, at tcp://localhost:5002
    bstar_t *bstar;
    if (argc == 2 && streq (argv [1], "-p")) {
        printf ("I: 主机master，等待备机（slave）连接。\n");
        bstar = bstar_new (BSTAR_PRIMARY,
            "tcp://*:5003", "tcp://localhost:5004");
        bstar_voter (bstar, "tcp://*:5001", ZMQ_ROUTER, s_echo, NULL);
    }
    else
    if (argc == 2 && streq (argv [1], "-b")) {
        printf ("I: 备机slave，等待主机（master）连接。\n");
        bstar = bstar_new (BSTAR_BACKUP,
            "tcp://*:5004", "tcp://localhost:5003");
        bstar_voter (bstar, "tcp://*:5002", ZMQ_ROUTER, s_echo, NULL);
    }
    else {
        printf ("Usage: bstarsrvs { -p | -b }\n");
        exit (0);
    }
    bstar_start (bstar);
    bstar_destroy (&bstar);
    return 0;
}