+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, size_t backlog_size)
+ : stream_id(stream_id),
+ data_fd(make_tempfile("")),
+ backlog_size(backlog_size),
+ bytes_received(0),
+ mark_pool(NULL)
+{
+ if (data_fd == -1) {
+ exit(1);
+ }
+}
+
+Stream::~Stream()
+{
+ if (data_fd != -1) {
+ int ret;
+ do {
+ ret = close(data_fd);
+ } while (ret == -1 && errno == EINTR);
+ if (ret == -1) {
+ perror("close");
+ }
+ }
+}
+
+Stream::Stream(const StreamProto &serialized)
+ : stream_id(serialized.stream_id()),
+ header(serialized.header()),
+ data_fd(make_tempfile(serialized.data())),
+ backlog_size(serialized.backlog_size()),
+ bytes_received(serialized.bytes_received()),
+ mark_pool(NULL)
+{
+ if (data_fd == -1) {
+ exit(1);
+ }
+}
+
+StreamProto Stream::serialize()
+{
+ StreamProto serialized;
+ serialized.set_header(header);
+ if (!read_tempfile(data_fd, serialized.mutable_data())) { // Closes data_fd.
+ exit(1);
+ }
+ serialized.set_backlog_size(backlog_size);
+ serialized.set_bytes_received(bytes_received);
+ serialized.set_stream_id(stream_id);
+ data_fd = -1;
+ 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();
+ }
+}
+