Skip to content

Commit fe950f1

Browse files
author
Simon Ninon
committed
initial commit with tcp client implementation (based on boost.asio)
1 parent 7767f89 commit fe950f1

6 files changed

Lines changed: 386 additions & 0 deletions

File tree

‎.gitignore‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -26,3 +26,5 @@
2626
*.exe
2727
*.out
2828
*.app
29+
30+
build

‎CMakeLists.txt‎

Lines changed: 70 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,70 @@
1+
###
2+
# config
3+
###
4+
cmake_minimum_required(VERSION 2.8.1)
5+
cmake_policy(SET CMP0042 NEW)
6+
7+
###
8+
# verbose make
9+
###
10+
# set(CMAKE_VERBOSE_MAKEFILE TRUE)
11+
12+
13+
###
14+
# project
15+
###
16+
set(PROJECT cpp_redis)
17+
project(${PROJECT} CXX)
18+
19+
20+
###
21+
# compilation options
22+
###
23+
set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -std=c++11")
24+
25+
26+
###
27+
# variables
28+
###
29+
set(CPP_REDIS_INCLUDES ${PROJECT_SOURCE_DIR}/includes)
30+
31+
32+
###
33+
# includes
34+
###
35+
include_directories(${CPP_REDIS_INCLUDES})
36+
37+
38+
###
39+
# link
40+
###
41+
link_directories(${GTEST_LIBS})
42+
43+
44+
###
45+
# sources
46+
###
47+
set(DIRS "sources" "sources/network")
48+
foreach(dir ${DIRS})
49+
# get directory sources
50+
file(GLOB s_${dir} "${dir}/*.cpp")
51+
# set sources
52+
set(SOURCES ${SOURCES} ${s_${dir}})
53+
endforeach()
54+
55+
56+
###
57+
# executable
58+
###
59+
add_library(${PROJECT} SHARED ${SOURCES})
60+
target_link_libraries(${PROJECT} boost_system)
61+
set_target_properties(${PROJECT}
62+
PROPERTIES
63+
LIBRARY_OUTPUT_DIRECTORY "${CMAKE_BINARY_DIR}/target")
64+
65+
###
66+
# examples
67+
###
68+
IF (BUILD_EXAMPLE)
69+
add_subdirectory(example)
70+
ENDIF(BUILD_EXAMPLE)

‎example/CMakeLists.txt‎

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,21 @@
1+
###
2+
# compilation options
3+
###
4+
set(CMAKE_CXX_FLAGS "${CMAKE_CXX_FLAGS} -std=c++11")
5+
6+
7+
###
8+
# includes
9+
###
10+
include_directories(${PROJECT_SOURCE_DIR}/includes
11+
${CPP_REDIS_INCLUDES})
12+
13+
14+
###
15+
# executable
16+
###
17+
add_executable(cpp_redis_example main.cpp)
18+
target_link_libraries(cpp_redis_example cpp_redis)
19+
set_target_properties(cpp_redis_example
20+
PROPERTIES
21+
RUNTIME_OUTPUT_DIRECTORY "${CMAKE_BINARY_DIR}/target")

‎example/main.cpp‎

Lines changed: 37 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,37 @@
1+
#include "cpp_redis/network/tcp_client.hpp"
2+
3+
#include <signal.h>
4+
#include <iostream>
5+
6+
bool should_exit = false;
7+
cpp_redis::network::tcp_client client;
8+
9+
void sigint_handler(int) {
10+
std::cout << "disconnected" << std::endl;
11+
client.disconnect();
12+
should_exit = true;
13+
}
14+
15+
int main(void) {
16+
client.set_receive_handler([] (cpp_redis::network::tcp_client&, const std::vector<char>& msg) {
17+
std::cout << "recv: " << std::string(msg.data()) << std::endl;
18+
client.send(msg);
19+
});
20+
21+
client.set_disconnection_handler([] (cpp_redis::network::tcp_client&) {
22+
std::cout << "disconnected" << std::endl;
23+
should_exit = true;
24+
});
25+
26+
if (not client.connect("0.0.0.0", 3000)) {
27+
std::cerr << "fail to connect" << std::endl;
28+
return -1;
29+
}
30+
31+
std::cout << "Connected" << std::endl;
32+
33+
signal(SIGINT, &sigint_handler);
34+
while (not should_exit);
35+
36+
return 0;
37+
}

‎includes/cpp_redis/network/tcp_client.hpp‎

Lines changed: 92 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,92 @@
1+
#pragma once
2+
3+
#include <mutex>
4+
#include <thread>
5+
#include <atomic>
6+
#include <stdexcept>
7+
#include <boost/asio.hpp>
8+
9+
using boost::asio::ip::tcp;
10+
11+
namespace cpp_redis {
12+
13+
namespace network {
14+
15+
//! tcp_client
16+
//! async tcp client based on boost asio
17+
class tcp_client {
18+
public:
19+
//! ctor & dtor
20+
tcp_client(void);
21+
~tcp_client(void);
22+
23+
//! assignment operator & copy ctor
24+
tcp_client(const tcp_client&) = delete;
25+
tcp_client& operator=(const tcp_client&) = delete;
26+
27+
//! returns whether the client is connected or not
28+
bool is_connected(void);
29+
30+
//! handle connection & disconnection
31+
bool connect(const std::string& host, unsigned int port);
32+
void disconnect(void);
33+
34+
//! send data
35+
void send(const std::string& buffer);
36+
void send(const std::vector<char>& buffer);
37+
38+
//! set receive callback
39+
//! called each time some data has been received
40+
typedef std::function<void(tcp_client&, const std::vector<char>& buffer)> receive_handler;
41+
void set_receive_handler(const receive_handler& handler);
42+
43+
//! set disconnection callback
44+
//! called each time a disconnection has been detected
45+
typedef std::function<void(tcp_client&)> disconnection_handler;
46+
void set_disconnection_handler(const disconnection_handler& handler);
47+
48+
public:
49+
//! exception handling
50+
class tcp_client_error : public std::runtime_error {
51+
public:
52+
using std::runtime_error::runtime_error;
53+
using std::runtime_error::what;
54+
};
55+
56+
private:
57+
//! make boost asio async read and write operations
58+
void async_read(void);
59+
void async_write(void);
60+
61+
//! close socket and call disconnect callback
62+
//! called in case of error
63+
void process_disconnection(void);
64+
65+
private:
66+
//! io service
67+
static boost::asio::io_service m_io_service;
68+
std::thread m_io_thread;
69+
tcp::socket m_socket;
70+
71+
//! is connected
72+
std::atomic_bool m_is_connected;
73+
74+
//! buffers
75+
static const unsigned int READ_SIZE = 2048;
76+
std::vector<char> m_read_buffer;
77+
std::vector<char> m_write_buffer;
78+
79+
//! handlers
80+
receive_handler m_receive_handler;
81+
disconnection_handler m_disconnection_handler;
82+
83+
//! thread safety
84+
std::mutex m_handler_mutex;
85+
std::mutex m_write_buffer_mutex;
86+
std::mutex m_receive_handler_mutex;
87+
std::mutex m_disconnection_handler_mutex;
88+
};
89+
90+
} //! network
91+
92+
} //! cpp_redis

‎sources/network/tcp_client.cpp‎

Lines changed: 164 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,164 @@
1+
#include "cpp_redis/network/tcp_client.hpp"
2+
3+
#include <iostream>
4+
5+
namespace cpp_redis {
6+
7+
namespace network {
8+
9+
boost::asio::io_service tcp_client::m_io_service;
10+
11+
tcp_client::tcp_client(void)
12+
: m_socket(m_io_service)
13+
, m_is_connected(false)
14+
, m_read_buffer(READ_SIZE) {}
15+
16+
tcp_client::~tcp_client(void) {
17+
if (m_io_thread.joinable())
18+
m_io_thread.join();
19+
}
20+
21+
bool
22+
tcp_client::connect(const std::string& host, unsigned int port) {
23+
if (m_is_connected)
24+
throw tcp_client_error("Already connected");
25+
26+
std::condition_variable conn_cond_var;
27+
28+
//! resolve host name
29+
boost::asio::ip::tcp::endpoint endpoint(boost::asio::ip::address::from_string(host), port);
30+
31+
//! async connect
32+
bool success;
33+
m_socket.async_connect(endpoint, [&](boost::system::error_code error) {
34+
success = not error;
35+
conn_cond_var.notify_one();
36+
37+
if (not error) {
38+
m_is_connected = true;
39+
async_read();
40+
}
41+
});
42+
43+
//! start loop
44+
m_io_thread = std::thread([this]() { m_io_service.run(); });
45+
46+
//! wait for async connect result
47+
std::mutex conn_mutex;
48+
std::unique_lock<std::mutex> lock(conn_mutex);
49+
conn_cond_var.wait(lock);
50+
51+
return success;
52+
}
53+
54+
void
55+
tcp_client::disconnect(void) {
56+
if (not m_is_connected)
57+
throw tcp_client_error("Not connected");
58+
59+
m_is_connected = false;
60+
m_io_service.post([this]() { m_socket.close(); });
61+
62+
if (m_io_thread.joinable())
63+
m_io_thread.join();
64+
}
65+
66+
void
67+
tcp_client::async_read(void) {
68+
boost::asio::async_read(m_socket, boost::asio::buffer(m_read_buffer.data(), READ_SIZE),
69+
[](const boost::system::error_code& error, std::size_t bytes) -> std::size_t {
70+
//! break if bytes have been received, continue otherwise
71+
return error or bytes ? 0 : READ_SIZE;
72+
},
73+
[=](boost::system::error_code error, std::size_t length) {
74+
if (error) {
75+
process_disconnection();
76+
return ;
77+
}
78+
79+
std::lock_guard<std::mutex> lock(m_receive_handler_mutex);
80+
if (m_receive_handler)
81+
m_receive_handler(*this, { m_read_buffer.begin(), m_read_buffer.begin() + length });
82+
83+
//! keep waiting for incoming bytes
84+
async_read();
85+
});
86+
}
87+
88+
void
89+
tcp_client::send(const std::string& buffer) {
90+
send(std::vector<char>{ buffer.begin(), buffer.end() });
91+
}
92+
93+
void
94+
tcp_client::send(const std::vector<char>& buffer) {
95+
if (not m_is_connected)
96+
throw tcp_client_error("Not connected");
97+
98+
if (not buffer.size())
99+
return ;
100+
101+
std::lock_guard<std::mutex> lock(m_write_buffer_mutex);
102+
103+
bool bytes_in_buffer = m_write_buffer.size() > 0;
104+
105+
//! concat buffer
106+
m_write_buffer.insert(m_write_buffer.end(), buffer.begin(), buffer.end());
107+
108+
//! if there were already bytes in buffer, simply return
109+
//! async_write callback will process the new buffer
110+
if (bytes_in_buffer)
111+
return;
112+
113+
async_write();
114+
}
115+
116+
void
117+
tcp_client::async_write(void) {
118+
boost::asio::async_write(m_socket, boost::asio::buffer(m_write_buffer.data(), m_write_buffer.size()),
119+
[this](boost::system::error_code error, std::size_t length) {
120+
if (error) {
121+
process_disconnection();
122+
return ;
123+
}
124+
125+
std::lock_guard<std::mutex> lock(m_write_buffer_mutex);
126+
m_write_buffer.erase(m_write_buffer.begin(), m_write_buffer.begin() + length);
127+
128+
if (m_write_buffer.size())
129+
async_write();
130+
});
131+
}
132+
133+
void
134+
tcp_client::process_disconnection(void) {
135+
m_is_connected = false;
136+
m_socket.close();
137+
138+
std::lock_guard<std::mutex> lock(m_disconnection_handler_mutex);
139+
if (m_disconnection_handler)
140+
m_disconnection_handler(*this);
141+
}
142+
143+
void
144+
tcp_client::set_receive_handler(const receive_handler& handler) {
145+
std::lock_guard<std::mutex> lock(m_receive_handler_mutex);
146+
147+
m_receive_handler = handler;
148+
}
149+
150+
void
151+
tcp_client::set_disconnection_handler(const disconnection_handler& handler) {
152+
std::lock_guard<std::mutex> lock(m_disconnection_handler_mutex);
153+
154+
m_disconnection_handler = handler;
155+
}
156+
157+
bool
158+
tcp_client::is_connected(void) {
159+
return m_is_connected;
160+
}
161+
162+
} //! network
163+
164+
} //! cpp_redis

0 commit comments

Comments
 (0)