+
+void Server::add_client_from_serialized(const ClientProto &client)
+{
+ MutexLock lock(&mutex);
+ Stream *stream;
+ map<string, Stream *>::iterator stream_it = streams.find(client.stream_id());
+ if (stream_it == streams.end()) {
+ stream = NULL;
+ } else {
+ stream = stream_it->second;
+ }
+ clients.insert(make_pair(client.sock(), Client(client, stream)));
+ Client *client_ptr = &clients[client.sock()];
+
+ // Start listening on data from this socket.
+ epoll_event ev;
+ if (client.state() == Client::READING_REQUEST) {
+ ev.events = EPOLLIN | EPOLLET | EPOLLRDHUP;
+ } else {
+ // If we don't have more data for this client, we'll be putting it into
+ // the sleeping array again soon.
+ ev.events = EPOLLOUT | EPOLLET | EPOLLRDHUP;
+ }
+ ev.data.u64 = 0; // Keep Valgrind happy.
+ ev.data.u64 = reinterpret_cast<uint64_t>(client_ptr);
+ if (epoll_ctl(epoll_fd, EPOLL_CTL_ADD, client.sock(), &ev) == -1) {
+ log_perror("epoll_ctl(EPOLL_CTL_ADD)");
+ exit(1);
+ }
+
+ if (client_ptr->state == Client::SENDING_DATA &&
+ client_ptr->stream_pos == client_ptr->stream->bytes_received) {
+ client_ptr->stream->put_client_to_sleep(client_ptr);
+ } else {
+ process_client(client_ptr);
+ }
+}
+
+void Server::add_stream(const string &stream_id, size_t backlog_size)
+{
+ MutexLock lock(&mutex);
+ streams.insert(make_pair(stream_id, new Stream(stream_id, backlog_size)));
+}
+
+void Server::add_stream_from_serialized(const StreamProto &stream)