+Client::Client(int sock)
+ : sock(sock),
+ fwmark(0),
+ connect_time(time(NULL)),
+ state(Client::READING_REQUEST),
+ stream(NULL),
+ header_or_error_bytes_sent(0),
+ bytes_sent(0)
+{
+ request.reserve(1024);
+
+ // Find the remote address, and convert it to ASCII.
+ sockaddr_in6 addr;
+ socklen_t addr_len = sizeof(addr);
+
+ if (getpeername(sock, reinterpret_cast<sockaddr *>(&addr), &addr_len) == -1) {
+ perror("getpeername");
+ remote_addr = "";
+ } else {
+ char buf[INET6_ADDRSTRLEN];
+ if (inet_ntop(addr.sin6_family, &addr.sin6_addr, buf, sizeof(buf)) == NULL) {
+ perror("inet_ntop");
+ remote_addr = "";
+ } else {
+ remote_addr = buf;
+ }
+ }
+}
+
+Client::Client(const ClientProto &serialized, Stream *stream)
+ : sock(serialized.sock()),
+ remote_addr(serialized.remote_addr()),
+ connect_time(serialized.connect_time()),
+ state(State(serialized.state())),
+ request(serialized.request()),
+ stream_id(serialized.stream_id()),
+ stream(stream),
+ header_or_error(serialized.header_or_error()),
+ header_or_error_bytes_sent(serialized.header_or_error_bytes_sent()),
+ bytes_sent(serialized.bytes_sent())
+{
+ if (stream->mark_pool != NULL) {
+ fwmark = stream->mark_pool->get_mark();
+ } else {
+ fwmark = 0; // No mark.
+ }
+ if (setsockopt(sock, SOL_SOCKET, SO_MARK, &fwmark, sizeof(fwmark)) == -1) {
+ if (fwmark != 0) {
+ perror("setsockopt(SO_MARK)");
+ }
+ }
+}
+
+ClientProto Client::serialize() const
+{
+ ClientProto serialized;
+ serialized.set_sock(sock);
+ serialized.set_remote_addr(remote_addr);
+ serialized.set_connect_time(connect_time);
+ serialized.set_state(state);
+ serialized.set_request(request);
+ serialized.set_stream_id(stream_id);
+ serialized.set_header_or_error(header_or_error);
+ serialized.set_header_or_error_bytes_sent(serialized.header_or_error_bytes_sent());
+ serialized.set_bytes_sent(bytes_sent);
+ return serialized;
+}
+
+ClientStats Client::get_stats() const
+{
+ ClientStats stats;
+ stats.stream_id = stream_id;
+ stats.remote_addr = remote_addr;
+ stats.connect_time = connect_time;
+ stats.bytes_sent = bytes_sent;
+ return stats;
+}
+
+Stream::Stream(const string &stream_id)
+ : stream_id(stream_id),
+ data(new char[BACKLOG_SIZE]),
+ data_size(0),
+ mark_pool(NULL)
+{
+ memset(data, 0, BACKLOG_SIZE);
+}
+
+Stream::~Stream()
+{
+ delete[] data;
+}
+
+Stream::Stream(const StreamProto &serialized)
+ : stream_id(serialized.stream_id()),
+ header(serialized.header()),
+ data(new char[BACKLOG_SIZE]),
+ data_size(serialized.data_size()),
+ mark_pool(NULL)
+{
+ assert(serialized.data().size() == BACKLOG_SIZE);
+ memcpy(data, serialized.data().data(), BACKLOG_SIZE);
+}
+
+StreamProto Stream::serialize() const
+{
+ StreamProto serialized;
+ serialized.set_header(header);
+ serialized.set_data(string(data, data + BACKLOG_SIZE));
+ serialized.set_data_size(data_size);
+ serialized.set_stream_id(stream_id);
+ return serialized;
+}
+
+void Stream::put_client_to_sleep(Client *client)
+{
+ sleeping_clients.push_back(client);
+}
+
+void Stream::wake_up_all_clients()
+{
+ if (to_process.empty()) {
+ swap(sleeping_clients, to_process);
+ } else {
+ to_process.insert(to_process.end(), sleeping_clients.begin(), sleeping_clients.end());
+ sleeping_clients.clear();
+ }
+}
+