int main (void) 
{
    void *context = zmq_ctx_new ();

    void *sink = zmq_socket (context, ZMQ_ROUTER);
    zmq_bind (sink, "inproc://example");

    //  First allow 0MQ to set the identity
    void *anonymous = zmq_socket (context, ZMQ_REQ);
    zmq_connect (anonymous, "inproc://example");
    s_send (anonymous, "ROUTER uses a generated UUID");
    s_dump (sink);

    //  Then set the identity ourselves
    void *identified = zmq_socket (context, ZMQ_REQ);
    zmq_setsockopt (identified, ZMQ_IDENTITY, "PEER2", 5);
    zmq_connect (identified, "inproc://example");
    s_send (identified, "ROUTER socket uses REQ's socket identity");
    s_dump (sink);

    zmq_close (sink);
    zmq_close (anonymous);
    zmq_close (identified);
    zmq_ctx_destroy (context);
    return 0;
}

//Process 0 is the leader process and will run this thread
void * sequencer(){

	int s = 1;
	int s_c[] = {1,1,1,1,1,1}; //vector timestamps for causal
	int sendTo, index;

	while(1){
		sendTo = s_request();
		if (ORDERING == TOTAL){
			s_send(s, sendTo);
			s++;
		}
		else if (ORDERING == CAUSAL){
			if (sendTo == -1)
				index = 0;
			else
				index = sendTo;
			//printf("%d\n",s_c[index]);
			s_send(s_c[index],sendTo);
			s_c[index]++;
		}
	}

	return 0;
}

int main (void) 
{
    void *context = zmq_init (1);

    void *sink = zmq_socket (context, ZMQ_XREP);
    zmq_bind (sink, "inproc://example");

    //  First allow 0MQ to set the identity
    void *anonymous = zmq_socket (context, ZMQ_REQ);
    zmq_connect (anonymous, "inproc://example");
    s_send (anonymous, "XREP uses a generated UUID");
    s_dump (sink);

    //  Then set the identity ourself
    void *identified = zmq_socket (context, ZMQ_REQ);
    zmq_setsockopt (identified, ZMQ_IDENTITY, "Hello", 5);
    zmq_connect (identified, "inproc://example");
    s_send (identified, "XREP socket uses REQ's socket identity");
    s_dump (sink);

    zmq_close (sink);
    zmq_close (anonymous);
    zmq_close (identified);
    zmq_term (context);
    return 0;
}

static void *
worker_task (void *args)
{
    void *context = zmq_ctx_new ();
    void *worker = zmq_socket (context, ZMQ_REQ);
    s_set_id (worker);          //  Set a printable identity
    zmq_connect (worker, "ipc://backend.ipc");

    //  Tell broker we're ready for work
    s_send (worker, "READY");

    while (1) {
        //  Read and save all frames until we get an empty frame
        //  In this example there is only 1 but it could be more
        char *identity = s_recv (worker);
        char *empty = s_recv (worker);
        assert (*empty == 0);
        free (empty);

        //  Get request, send reply
        char *request = s_recv (worker);
        printf ("Worker: %s\n", request);
        free (request);

        s_sendmore (worker, identity);
        s_sendmore (worker, "");
        s_send     (worker, "OK");
        free (identity);
    }
    zmq_close (worker);
    zmq_ctx_destroy (context);
    return NULL;
}

int main () {
    zmq::context_t context(1);

    //  Subscriber tells us when it's ready here
    zmq::socket_t sync(context, ZMQ_PULL);
    sync.bind("tcp://*:5564");

    //  We send updates via this socket
    zmq::socket_t publisher (context, ZMQ_PUB);
    publisher.bind("tcp://*:5565");

    //  Wait for synchronization request
    s_recv (sync);

    //  Now broadcast exactly 10 updates with pause
    int update_nbr;
    for (update_nbr = 0; update_nbr < 10; update_nbr++) {
       
        std::ostringstream oss;
        oss << "Update "<< update_nbr ;
        s_send (publisher, oss.str());
        sleep (1);
    }
    s_send (publisher, "END");

    sleep (1);              //  Give 0MQ/2.0.x time to flush output
    return 0;
}

int main (void) 
{
    s_version_assert (2, 1);
    void *context = zmq_init (1);

    //  Socket to talk to clients
    void *publisher = zmq_socket (context, ZMQ_PUB);
    zmq_bind (publisher, "tcp://*:5561");

    //  Socket to receive signals
    void *syncservice = zmq_socket (context, ZMQ_REP);
    zmq_bind (syncservice, "tcp://*:5562");

    //  Get synchronization from subscribers
    int subscribers = 0;
    while (subscribers < SUBSCRIBERS_EXPECTED) {
        //  - wait for synchronization request
        char *string = s_recv (syncservice);
        free (string);
        //  - send synchronization reply
        s_send (syncservice, "");
        subscribers++;
    }
    //  Now broadcast exactly 1M updates followed by END
    int update_nbr;
    for (update_nbr = 0; update_nbr < 1000000; update_nbr++)
        s_send (publisher, "Rhubarb");

    s_send (publisher, "END");

    zmq_close (publisher);
    zmq_close (syncservice);
    zmq_term (context);
    return 0;
}

//  Worker using REQ socket to do LRU routing
//
static void *
worker_thread (void *args) {
    void *context = zmq_init (1);
    void *worker = zmq_socket (context, ZMQ_REQ);
    s_set_id (worker);          //  Makes tracing easier
    zmq_connect (worker, "ipc://backend.ipc");

    //  Tell broker we're ready for work
    s_send (worker, "READY");

    while (1) {
        //  Read and save all frames until we get an empty frame
        //  In this example there is only 1 but it could be more
        char *address = s_recv (worker);
        char *empty = s_recv (worker);
        assert (*empty == 0);
        free (empty);

        //  Get request, send reply
        char *request = s_recv (worker);
        printf ("Worker: %s\n", request);
        free (request);

        s_sendmore (worker, address);
        s_sendmore (worker, "");
        s_send     (worker, "OK");
        free (address);
    }
    zmq_close (worker);
    zmq_term (context);
    return NULL;
}

//  We will do this all in one thread to emphasize the sequence
//  of events...
int main () {
    zmq::context_t context(1);

    zmq::socket_t client (context, ZMQ_XREP);
   client.bind("ipc://routing.ipc");

    zmq::socket_t worker (context, ZMQ_REP);
    worker.setsockopt(ZMQ_IDENTITY, "A", 1);
    worker.connect("ipc://routing.ipc");

    //  Wait for sockets to stabilize
    sleep (1);

    //  Send papa address, address stack, empty part, and request
    s_sendmore (client, "A");
    s_sendmore (client, "address 3");
    s_sendmore (client, "address 2");
    s_sendmore (client, "address 1");
    s_sendmore (client, "");
    s_send     (client, "This is the workload");

    //  Worker should get just the workload
    s_dump (worker);

    //  We don't play with envelopes in the worker
    s_send (worker, "This is the reply");

    //  Now dump what we got off the XREP socket...
    s_dump (client);

    return 0;
}

//recRequest by one of the Satellites, reply with number of stored msgs and the oldest stored msg
void ReefServer::recRequest(std::string aka){
	//find satellite by alias in the map
	auto search = satMsgControlMap.find(aka);

	if (search != satMsgControlMap.end()) { //if sat has been found
		//get the Controlnumbers for this satellite
		satelliteMsgControl& msgControl = search->second;
		int msgCount = msgControl.control[1]; //number of stored Messages for this Satellite

		if (!msgCount){
			s_send(rep, std::to_string(msgCount)); //send the number of messages to the satellite
		}
		else{
			s_sendmore(rep, std::to_string(msgCount)); //send the number of messages to the satellite			
			int msgPosition = msgControl.control[0]; //find position of Message Queue in the vector of all Queues

			std::pair<std::string, std::string> msg = satelliteMsgs[msgPosition][0];

			s_sendmore(rep, msg.first); //send the Message to the satellite
			s_send(rep, msg.second);

			satelliteMsgs[msgPosition].erase(satelliteMsgs[msgPosition].begin()); //erase the Message from the Queue
			msgControl.control[1]--; //lower the count of stored Messages by 1	
			CUR_MESSAGES--;
		}
	}
	else{
		s_send(rep, "-1");
	}


}

//  We will do this all in one thread to emphasize the sequence
//  of events...
int main () {
    void *context = zmq_init (1);

    void *client = zmq_socket (context, ZMQ_XREP);
    zmq_bind (client, "ipc://routing");

    void *worker = zmq_socket (context, ZMQ_REP);
    zmq_setsockopt (worker, ZMQ_IDENTITY, "A", 1);
    zmq_connect (worker, "ipc://routing");

    //  Wait for sockets to stabilize
    sleep (1);

    //  Send papa address, address stack, empty part, and request
    s_sendmore (client, "A");
    s_sendmore (client, "address 3");
    s_sendmore (client, "address 2");
    s_sendmore (client, "address 1");
    s_sendmore (client, "");
    s_send     (client, "This is the workload");

    //  Worker should get just the workload
    s_dump (worker);

    //  We don't play with envelopes in the worker
    s_send (worker, "This is the reply");

    //  Now dump what we got off the XREP socket...
    s_dump (client);

    zmq_term (context);
    return 0;
}

//  Worker using REQ socket to do LRU routing
//
static void *
worker_thread (void *arg) {
	zmq::context_t context(1);
    zmq::socket_t worker (context, ZMQ_REQ);
    s_set_id (worker);          //  Makes tracing easier
    worker.connect("ipc://backend.ipc");

    //  Tell backend we're ready for work
    s_send (worker, "READY");

    while (1) {
        //  Read and save all frames until we get an empty frame
        //  In this example there is only 1 but it could be more
        std::string address = s_recv (worker);
        {
            std::string empty = s_recv (worker);
            assert (empty.size() == 0);
        }

        //  Get request, send reply
        std::string request = s_recv (worker);
        std::cout << "Worker: " << request << std::endl;

        s_sendmore (worker, address);
        s_sendmore (worker, "");
        s_send     (worker, "OK");
    }
    return (NULL);
}

static void
zap_handler (void *ctx)
{
    //  Create and bind ZAP socket
    void *zap = zmq_socket (ctx, ZMQ_REP);
    assert (zap);
    int rc = zmq_bind (zap, "inproc://zeromq.zap.01");
    assert (rc == 0);

    //  Process ZAP requests forever
    while (true) {
        char *version = s_recv (zap);
        if (!version)
            break;          //  Terminating
        char *sequence = s_recv (zap);
        char *domain = s_recv (zap);
        char *address = s_recv (zap);
        char *identity = s_recv (zap);
        char *mechanism = s_recv (zap);
        char *username = s_recv (zap);
        char *password = s_recv (zap);

        assert (streq (version, "1.0"));
        assert (streq (mechanism, "PLAIN"));
        assert (streq (identity, "IDENT"));

        s_sendmore (zap, version);
        s_sendmore (zap, sequence);
        if (streq (username, "admin")
        &&  streq (password, "password")) {
            s_sendmore (zap, "200");
            s_sendmore (zap, "OK");
            s_sendmore (zap, "anonymous");
            s_send (zap, "");
        }
        else {
            s_sendmore (zap, "400");
            s_sendmore (zap, "Invalid username or password");
            s_sendmore (zap, "");
            s_send (zap, "");
        }
        free (version);
        free (sequence);
        free (domain);
        free (address);
        free (identity);
        free (mechanism);
        free (username);
        free (password);
    }
    rc = zmq_close (zap);
    assert (rc == 0);
}

int main(int argc, char * argv[])
{
	if(argc < 2)
	{
		cout<<"please enter a file name\n";
		exit(-1);
	}
	char * filename = argv[1];
	FILE * comfile = fopen(filename,"r");
	if(!comfile)
	{
		perror("failed to open command file");
		exit(-1);
	}

	map<int,int> ports = parseMembers();
	map<int,int> nodeToSocket;
	
	map<int, int>::const_iterator endports = ports.end();
	for(map<int, int>::const_iterator it = ports.begin(); it != endports; it++) 
	{
		nodeToSocket[it->first] = new_socket();
		connect(nodeToSocket[it->first],it->second);
	}
	ports.clear();
	s_send(nodeToSocket[0],"-1");
	
	char com[BUFFER_SIZE];
	char tmp[BUFFER_SIZE];
	while(fgets(com, BUFFER_SIZE,comfile) != NULL){
		strcpy(tmp,com);
		unsigned int sleeptime = atoi(strtok(com,"\n:"));
		sleeptime = sleeptime * 1000;
		usleep(sleeptime);
		int nodeID = atoi(strtok(NULL,"\n:"));

		//locate the 2nd colon for commandID options
		int i,j;
		for(i = 0,j = 0; j < 2 && i < BUFFER_SIZE; i++){
			if(tmp[i] == ':')
				j++;
		}

		char * comAndOpt = tmp + i;
		char moretmp[BUFFER_SIZE];
		strcpy(moretmp,"0:C:");
		strcat(moretmp,strtok(comAndOpt,"\n")); //strtok strips new line from command
		//send comAndOpt to nodeID
		printf("sending %s to %d\n",moretmp,nodeID);
		s_send(nodeToSocket[nodeID],moretmp);
	}
	return 0;
}

int main () {
    zmq::context_t context (1);

    zmq::socket_t * client = s_client_socket (context);

    int sequence = 0;
    int retries_left = REQUEST_RETRIES;

    while (retries_left) {
        std::stringstream request;
        request << ++sequence;
        s_send (*client, request.str());
        sleep (1);

        bool expect_reply = true;
        while (expect_reply) {
            //  Poll socket for a reply, with timeout
            zmq::pollitem_t items[] = { { *client, 0, ZMQ_POLLIN, 0 } };
            zmq::poll (&items[0], 1, REQUEST_TIMEOUT * 1000);

            //  If we got a reply, process it
            if (items[0].revents & ZMQ_POLLIN) {
                //  We got a reply from the server, must match sequence
                std::string reply = s_recv (*client);
                if (atoi (reply.c_str ()) == sequence) {
                    std::cout << "I: server replied OK (" << reply << ")" << std::endl;
                    retries_left = REQUEST_RETRIES;
                    expect_reply = false;
                }
                else {
                    std::cout << "E: malformed reply from server: " << reply << std::endl;
                }
            }
            else
            if (--retries_left == 0) {
                std::cout << "E: server seems to be offline, abandoning" << std::endl;
                expect_reply = false;
                break;
            }
            else {
                std::cout << "W: no response from server, retrying..." << std::endl;
                //  Old socket will be confused; close it and open a new one
                delete client;
                client = s_client_socket (context);
                //  Send request again, on new socket
                s_send (*client, request.str());
            }
        }
    }
    delete client;
    return 0;
}

int main(int argc, char *argv[])
{
    void *ctx, *handler, *thread, *server, *client;
    const char *domain, *connect_addr;
    int optval;

    domain = argc > 1 ? argv[1] : "test";
    connect_addr = strcmp(domain, "fail") ? "tcp://127.0.0.1:9000" : "tcp://127.0.0.1:9001";

    assert((ctx = zmq_ctx_new()));

    /* Start ZAP handler thread. */
    assert((handler = zmq_socket(ctx, ZMQ_REP)));
    assert(!(zmq_bind(handler, "inproc://zeromq.zap.01")));
    assert((thread = zmq_threadstart(zap_handler, handler)));

    /* Bind server. */
    assert((server = zmq_socket(ctx, ZMQ_DEALER)));
    assert(!(zmq_setsockopt(server, ZMQ_ZAP_DOMAIN, domain, strlen(domain))));
    assert(!zmq_bind(server, "tcp://127.0.0.1:9000"));

    /* Connect client. */
    assert((client = zmq_socket(ctx, ZMQ_DEALER)));
    optval = 200;
    assert(!(zmq_setsockopt(client, ZMQ_RECONNECT_IVL, &optval, sizeof(optval))));
    optval = 5000;
    assert(!(zmq_setsockopt(client, ZMQ_RECONNECT_IVL_MAX, &optval, sizeof(optval))));
    optval = 30000;
    assert(!(zmq_setsockopt(client, ZMQ_SNDTIMEO, &optval, sizeof(optval))));
    optval = 1;
    assert(!(zmq_setsockopt(client, ZMQ_IMMEDIATE, &optval, sizeof(optval))));
    assert(!zmq_connect(client, connect_addr));

    /* Bounce test. */
    s_send(client, "Hello, Server!");
    assert(!strcmp(s_recv(server), "Hello, Server!"));
    s_send(server, "Hello, Client!");
    assert(!strcmp(s_recv(client), "Hello, Client!"));

    /* Cleanup. */
    assert(!zmq_disconnect(client, connect_addr));
    assert(!zmq_close(client));

    assert(!zmq_unbind(server, "tcp://127.0.0.1:9000"));
    assert(!zmq_close(server));

    assert(!zmq_term(ctx));
    zmq_threadclose(thread);

    return 0;
}

static void zap_handler (void *handler)
{
    //  Process ZAP requests forever
    while (true) {
        char *version = s_recv (handler);
        if (!version)
            break;          //  Terminating

        char *sequence = s_recv (handler);
        char *domain = s_recv (handler);
        char *address = s_recv (handler);
        char *identity = s_recv (handler);
        char *mechanism = s_recv (handler);
        uint8_t client_key [32];
        int size = zmq_recv (handler, client_key, 32, 0);
        assert (size == 32);

        char client_key_text [40];
        zmq_z85_encode (client_key_text, client_key, 32);

        assert (streq (version, "1.0"));
        assert (streq (mechanism, "CURVE"));
        assert (streq (identity, "IDENT"));

        s_sendmore (handler, version);
        s_sendmore (handler, sequence);

        if (streq (client_key_text, client_public)) {
            s_sendmore (handler, "200");
            s_sendmore (handler, "OK");
            s_sendmore (handler, "anonymous");
            s_send     (handler, "");
        }
        else {
            s_sendmore (handler, "400");
            s_sendmore (handler, "Invalid client public key");
            s_sendmore (handler, "");
            s_send     (handler, "");
        }
        free (version);
        free (sequence);
        free (domain);
        free (address);
        free (identity);
        free (mechanism);
    }
    zmq_close (handler);
}

static void *
worker_thread (void *arg) {

    zmq::context_t * context = (zmq::context_t *)arg;
    zmq::socket_t worker (*context, ZMQ_REQ);
    
    //  We use a string identity for ease here
    s_set_id (worker);
    worker.connect("ipc://routing.ipc");

    int total = 0;
    while (1) {
        //  Tell the router we're ready for work
        s_send (worker, "ready");

        //  Get workload from router, until finished
        std::string workload = s_recv (worker);
        int finished = (workload.compare("END") == 0);
        
        if (finished) {
            std::cout << "Processed: " << total << " tasks" << std::endl;
            break;
        }
        total++;

        //  Do some random work
        s_sleep(within (100) + 1);
    }
    return (NULL);
}

void replay(void *arg) {
	void *slave = zmq_socket(arg, ZMQ_REQ);
	/*
	if (done % 2) {
		zmq_connect (slave, "tcp://localhost:5562");
	}
	else {
		zmq_connect (slave, "tcp://localhost:5563");
	}*/
	zmq_connect (slave, "tcp://localhost:5562");

	char *string;
	char buf[1024];
	sprintf(buf, "%d", rand());
	s_send(slave, buf);
	string = s_recv(slave);
	if (strcmp(string , buf) != 0) {
		debug_puts("ERROR");
	}
	free(string);
    pthread_mutex_lock(&lock);
    done++;
    pthread_mutex_unlock(&lock);
	zmq_close(slave);
}

int main(int argc, char **argv)
{
    threadpool_t *pool;
	int i, k, ret=0;
	void *slave[SLAVE_NUM];
	char *string;

	void *context = zmq_init(1);


    pthread_mutex_init(&lock, NULL);

    pool = threadpool_init(THREAD, QUEUE, 0);
    fprintf(stderr, "Pool started with %d threads and "
            "queue size of %d\n", THREAD, QUEUE);
	if (!pool) {
		return -1;
	}

	for (i = 0; i < 10; i++) {
		/*k = i % 2;
		if (k!=0 && k!=1) {
			ret = -100;
			break;
		}*/
    	ret = threadpool_add(pool, &replay, context, 0);
		if (ret) {
			break;
		}
       	pthread_mutex_lock(&lock);
        tasks++;
        pthread_mutex_unlock(&lock);
	}

    fprintf(stderr, "RET: %d Added %d tasks\n", ret, tasks);

    while(tasks / 2 > done) {
        sleep(1);
    }
    fprintf(stderr, "Did %d tasks before shutdown\n", done);
    fprintf(stderr, "Did %d tasks\n", done);

	slave[0] = zmq_socket (context, ZMQ_REQ);
	zmq_connect (slave[0], "tcp://localhost:5562");
/*
	slave[1] = zmq_socket (context, ZMQ_REQ);
	zmq_connect (slave[1], "tcp://localhost:5563");
*/
	for (i = 0; i < SLAVE_NUM; i++) {
		s_send(slave[i], "END");
		string = s_recv(slave[i]);
		free(string);
		zmq_close(slave[i]);
	}

	zmq_term(context);
    threadpool_destroy(pool, 0);

    return 0;
}

int
query_users (void *ctx, char *filter)
{
	assert(ctx);
	assert(filter);

	console_log("Sending query users request...\n");

	void *sock = zmq_socket(ctx, ZMQ_REQ);
	zmq_connect(sock, "tcp://127.0.0.1:5555");

	char cmd_msg[250];	//$$ Longer than length will crash
	sprintf(cmd_msg, "users\n%s", filter);
	int num_bytes_sent = s_send(sock, cmd_msg);
	if (num_bytes_sent == -1) {
		console_log("Error sending query users request\n");

		zmq_close(sock);
		return -1;
	}

	console_log("Waiting for query users reply...\n");
	char *reply_msg = s_recv(sock);
	if (reply_msg != NULL) {
		console_log("Received reply: '%s'\n", reply_msg);
		free(reply_msg);
	}

	zmq_close(sock);
	return 0;
}