void test_bitcount(client &c) {
    c.flushdb();
    c.set("mykey", "foobar");
    assert(c.bitcount("mykey")==26);
    assert(c.bitcount("mykey", {0,0})==4);
    assert(c.bitcount("mykey", {1,1})==6);
    assert(c.bitcount("nonexist", {8,12})==0);
}

void echoMessage(int i)
{
    echoed = false;

    clientIO.send(heavyConnections[i], ECHO_MESSAGE, websocketpp::frame::opcode::value::TEXT);

    while(!echoed) {
        clientIO.run_one();
    }
}

void nextConnection()
{
    static std::string uri;

    if (connections % CONNECTIONS_PER_INTERFACE == 0) {
        uri = "ws://127.0.0." + to_string(connections / CONNECTIONS_PER_INTERFACE + 1) + ":" + to_string(port);
    }

    websocketpp::lib::error_code ec;
    connection_ptr con = clientIO.get_connection(uri, ec);
    clientIO.connect(con);
}

int main(int argc, char **argv)
{
    clientIO.clear_access_channels(websocketpp::log::alevel::all);
    clientIO.init_asio();

    clientIO.set_message_handler([](websocketpp::connection_hdl hdl, client::message_ptr msg) {
        if (ECHO_MESSAGE != msg->get_payload()) {
            cout << "Error echo" << endl;
            exit(-1);
        }
        echoed = true;
    });

    clientIO.set_open_handler([&clientIO](websocketpp::connection_hdl hdl) {
        connected = true;
        heavyConnections.push_back(hdl);
    });

    startPoint = chrono::system_clock::now();
    while(connections < TOTAL_CONNECTIONS) {

        // Idle connections are made very lightweight
        for (int i = 0; i < IDLE_CONNECTIONS_PER_ACTIVE_CONNECTION
             && connections < TOTAL_CONNECTIONS; i++) {
            if (nextLightweightConnection()) {
                return -1;
            }
        }

        // Active connections are "real" using WebSocket++
        nextHeavyweightConnection();
    }

    cout << "----------------------------------" << endl;

    cout << "Active connections: " << heavyConnections.size() << endl;
    cout << "Idle connections: " << (connections - heavyConnections.size()) << endl;
    float connectionsPerSecond = float(connections) / chrono::duration_cast<chrono::milliseconds>(chrono::system_clock::now() - startPoint).count();
    cout << "Connection performance: " << connectionsPerSecond << " connections/ms" << endl;

    // start sending messages using the active connections
    cout << "----------------------------------" << endl;
    startPoint = chrono::system_clock::now();
    for (int i = 0; i < heavyConnections.size(); i++) {
        echoMessage(i);
    }
    cout << "Echo delay, microseconds: " << chrono::duration_cast<chrono::microseconds>(chrono::system_clock::now() - startPoint).count() << endl;

    cout << "ALL DONE" << endl;
    clientIO.run();

    return 0;
}

void run_time_limited_client(client & c, std::string uri, long timeout,
    bool log)
{
    if (log) {
        c.set_access_channels(websocketpp::log::alevel::all);
        c.set_error_channels(websocketpp::log::elevel::all);
    } else {
        c.clear_access_channels(websocketpp::log::alevel::all);
        c.clear_error_channels(websocketpp::log::elevel::all);
    }
    c.init_asio();

    websocketpp::lib::error_code ec;
    client::connection_ptr con = c.get_connection(uri,ec);
    BOOST_CHECK( !ec );
    c.connect(con);

    websocketpp::lib::thread tthread(websocketpp::lib::bind(
        &close_after_timeout<client>,
        websocketpp::lib::ref(c),
        con->get_handle(),
        timeout
    ));
    tthread.detach();

    c.run();
}

void nextHeavyweightConnection()
{
    connected = false;
    string uri = "ws://127.0.0." + to_string(address) + ":" + to_string(PORT);
    websocketpp::lib::error_code ec;
    connection_ptr con = clientIO.get_connection(uri, ec);
    clientIO.connect(con);

    while(!connected) {
        clientIO.run_one();
    }

    updateAddress();
}

void test_sub(client &c) {
    message_queue &q=c.subscribe({"abc", "def"});
    fiber pubber(test_pub);
    redis_message msg;
    q.pop(msg);
    assert(msg.first=="abc");
    assert(msg.second=="123");
    q.pop(msg);
    assert(msg.first=="def");
    assert(msg.second=="456");
    c.unsubscribe({"abc", "def"});
    this_fiber::sleep_for(std::chrono::seconds(1));
    assert(!q.is_open());
    pubber.join();
}

void test_get_set(client &c) {
    c.flushdb();
    c.set("k1", "v1");
    assert(*c.get("k1")=="v1");
    c.set("k1", "v2");
    assert(*c.get("k1")=="v2");
    assert(!c.get("nonexist"));

    c.set("mykey", "10.5");
    c.incrbyfloat("mykey", "0.1");
    assert(*(c.get("mykey"))=="10.6");
}

inline error_code async_submit_job(client const &client, char const *rsl, JobSubmitCallback const &job_submit_callback, int state_mask, StateChangeCallback const &state_chage_callback)
{
	typedef listen_state_op<StateChangeCallback, non_blocking> listen_state_op_type;
	listen_state_op_type *listener_op = new listen_state_op_type(state_chage_callback);

	char *callback_contact(0);
	int const callback_allowed(
		::globus_gram_client_callback_allow(
			&listen_state_op<StateChangeCallback, non_blocking>::callback,
			listener_op, &callback_contact));
	if (callback_allowed != GLOBUS_SUCCESS)
	{
		delete listener_op;
		return static_cast<error_code>(callback_allowed);
	}

	job_submit_op<JobSubmitCallback> *op = new job_submit_op<JobSubmitCallback>(job_submit_callback);
	int const job_requested(
		::globus_gram_client_register_job_request(
			client.contact(), rsl, state_mask, callback_contact, 0, &job_submit_op<JobSubmitCallback>::callback, op));
	if (job_requested != GLOBUS_SUCCESS)
	{
		delete listener_op;
		delete op;
		return static_cast<error_code>(job_requested);
	}
	return no_error;
}

inline error_code submit_job_wait_until(client const &client, char const *rsl, int state_mask, StateChangeListener const &listener)
{
	listen_state_op<StateChangeListener, blocking> listener_op(listener);

	char *callback_contact(0);
	int const callback_allowed(
		::globus_gram_client_callback_allow(
			&listen_state_op<StateChangeListener, blocking>::callback,
			&listener_op, &callback_contact));
	if (callback_allowed != GLOBUS_SUCCESS) return static_cast<error_code>(callback_allowed);
	listener_op.listen();

	char *job_contact(0);
	int const job_requested(
		::globus_gram_client_job_request(
			client.contact(), rsl, state_mask, callback_contact, &job_contact));
	if (job_requested != GLOBUS_SUCCESS)
	{
		listener_op.cancel();
		return static_cast<error_code>(job_requested);
	}

	typedef std::pointer_to_unary_function<void *,void> function_type;
	globus::util::on_exit<function_type> free_contact(
		std::ptr_fun(::free), job_contact);

	return no_error;
}

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

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

void test_sscan(client &c) {
    c.flushdb();
    std::set<std::string> expected;
    for (int i=42; i<442; i++) {
        std::string k("key");
        k+=boost::lexical_cast<std::string>(i);
        c.sadd("myset", {k});
        expected.insert(k);
    }

    std::set<std::string> keys;
    for(auto i=c.sscan("myset"); i!=c.end(); ++i) {
        keys.insert(*i);
    }
    assert(keys==expected);
}

void test_zscan(client &c) {
    c.flushdb();
    std::set<std::string> expected;
    for (int i=42; i<442; i++) {
        std::string k("key");
        k+=boost::lexical_cast<std::string>(i);
        c.zadd("myzset", {{1.2, k}});
        expected.insert(k);
    }

    std::set<std::string> keys;
    for(auto i=c.zscan("myzset"); i!=c.end(); ++i) {
        keys.insert(*i);
        assert(i.score()==1.2);
    }
    assert(keys==expected);
}

void test_hscan(client &c) {
    c.flushdb();
    std::set<std::string> expected;
    for (int i=42; i<442; i++) {
        std::string k("key");
        k+=boost::lexical_cast<std::string>(i);
        std::string v("key");
        v+=boost::lexical_cast<std::string>(i);
        c.hset("myhash", k, v);
        expected.insert(k);
    }

    std::set<std::string> keys;
    for(auto i=c.hscan("myhash"); i!=c.end(); ++i) {
        keys.insert(*i);
    }
    assert(keys==expected);
}

void on_open(client* c, websocketpp::connection_hdl hdl) {
    // now it is safe to use the connection
    std::cout << "connection ready" << std::endl;

    received=false;
    // Send a SIP OPTIONS message to the server:
    std::string SIP_msg="OPTIONS sip:[email protected] SIP/2.0\r\nVia: SIP/2.0/WS df7jal23ls0d.invalid;rport;branch=z9hG4bKhjhs8ass877\r\nMax-Forwards: 70\r\nTo: <sip:car[email protected]>\r\nFrom: Alice <sip:[email protected]>;tag=1928301774\r\nCall-ID: a84b4c76e66710\r\nCSeq: 63104 OPTIONS\r\nContact: <sip:[email protected]>\r\nAccept: application/sdp\r\nContent-Length: 0\r\n\r\n";
    sip_client.send(hdl, SIP_msg.c_str(), websocketpp::frame::opcode::text);
}

void echo(int messages, int factor)
{
    for (int j = 0; j < factor; j++) {
        for (int i = 0; i < messages; i++) {
            clientIO.send(connectionVector[rand() % connectionVector.size()], ECHO_MESSAGE, websocketpp::frame::opcode::value::BINARY);
            sent++;
        }
    }
}

int main()
{
	// Remove superfluous logging
	test_client.clear_access_channels(websocketpp::log::alevel::all);
    test_client.clear_error_channels(websocketpp::log::elevel::all);
        
	// Normal endpoint setup (just as you would with the regular websocketpp::client)
	test_client.init_asio();
	// The endpoint must be perpetual. TODO look at supporting non perpetual (will have to use .reset())
	test_client.start_perpetual();
	// Start spinning the thread
	test_thread.reset(new websocketpp::lib::thread(&client::run, &test_client));
	
	// Done boilerplate initialization, now our connection code:
	websocketpp::lib::error_code ec;
	client::connection_ptr con = test_client.get_connection("ws://some.fake.server.that.should.not.exist:9001", ec);
	if(ec)
	{
		throw ec;
	}
	// Setup any retry_data settings here
	con->m_retry = true;		// Indicates that we do want to attempt to retry connecting (if first attempt fails)
	con->m_retry_delay = 300;	// Will wait 300ms between attempts
	con->m_max_attempts = 10;	// Will stop attempting to retry after 10 attempts
	
	try
	{
		// Delibrately call connect without setting a configure_handler
		// This shows the importance that when setting handlers such as
		// open, message, closed... handlers it must be done within the
		// configure handler because each retry attempt creates a new
		// connection using get_connection(...)
		test_client.connect(con);
    } catch (const websocketpp::exception & e) {
        std::cout << e.what() << std::endl;
    }
    
    // Must setup a configure handler, where we register specific connection items (eg. handlers, etc)...
	con->set_configure_handler(bind(&configure_con, &test_client, ::_1));
    
    std::cout << "Sleeping for 4 seconds to simulate a server that we cannot connect to" << std::endl;
    std::cout << "and 4 seconds is enough for the 10 retries (@ 300ms delay) to run it's course" << std::endl;
    
    // Now connect will start attempting to connect
    test_client.connect(con);
    
    std::this_thread::sleep_for(std::chrono::seconds(6));
    
    test_client.stop_perpetual();
    test_thread->join();
    return 0;
}

void test_sort(client &c) {
    c.flushdb();
    c.lpush("mylist", {"c"});
    c.lpush("mylist", {"b", "a"});
    assert((c.sort("mylist", client::sort_criteria().alpha())==std::list<std::string> {"a", "b", "c"}));
    assert((c.sort("mylist", client::sort_criteria().alpha().desc())==std::list<std::string> {"c", "b", "a"}));
    assert(c.sort("mylist", client::sort_criteria().alpha().desc(), "mylist2")==3);
    assert((c.lrange("mylist2", 0, 100)==std::list<std::string> {"c", "b", "a"}));
}

void client::onPing(client& c, std::string message)
{
    std::smatch match;
    std::regex e("PING :(.*)");
    std::regex_match(message, match, e);

    if(match.size())
    {
        std::string pong("PONG :");
        pong += match[1];
        c.send(pong);
    }
}

    void operator()(client& c, std::string message)
    {
        std::smatch match;
        std::regex expression("^:(\\S+) PRIVMSG (\\S+)+ :!alice$");
        std::regex_search(message, match, expression);

        if(match.size())
        {
            std::string sender = match[2];
            std::string message = g.getSentence();
            c.send("PRIVMSG " + sender + " :" + message);
        }
    }