X-Git-Url: https://git.sesse.net/?p=cubemap;a=blobdiff_plain;f=main.cpp;h=0aa91f38becb8ab9b0d9a6cd75a6a11278e997d0;hp=fa63549bd7f8a00b1ce05d55ad09c5f594507cfa;hb=5605c88a27fcf64ea8f6fd6cbb52dada73856ee0;hpb=340489a8e732519182ecbc92116e7dfa2997143c

diff --git a/main.cpp b/main.cpp
index fa63549..0aa91f3 100644
--- a/main.cpp
+++ b/main.cpp
@@ -16,8 +16,8 @@
 #include <utility>
 #include <vector>
 
-#include "accesslog.h"
 #include "acceptor.h"
+#include "accesslog.h"
 #include "config.h"
 #include "input.h"
 #include "log.h"
@@ -25,6 +25,7 @@
 #include "serverpool.h"
 #include "state.pb.h"
 #include "stats.h"
+#include "stream.h"
 #include "util.h"
 #include "version.h"
 
@@ -49,6 +50,10 @@ void hup(int signum)
 	}
 }
 
+void do_nothing(int signum)
+{
+}
+
 CubemapStateProto collect_state(const timeval &serialize_start,
                                 const vector<Acceptor *> acceptors,
                                 const multimap<string, InputWithRefcount> inputs,
@@ -143,10 +148,14 @@ void create_streams(const Config &config,
 	for (unsigned i = 0; i < config.streams.size(); ++i) {
 		const StreamConfig &stream_config = config.streams[i];
 		if (deserialized_stream_ids.count(stream_config.stream_id) == 0) {
-			servers->add_stream(stream_config.stream_id, stream_config.backlog_size);
+			servers->add_stream(stream_config.stream_id,
+			                    stream_config.backlog_size,
+			                    Stream::Encoding(stream_config.encoding));
 		} else {
 			servers->set_backlog_size(stream_config.stream_id, stream_config.backlog_size);
 		}
+		servers->set_encoding(stream_config.stream_id,
+				      Stream::Encoding(stream_config.encoding));
 		expecting_stream_ids.erase(stream_config.stream_id);
 
 		if (stream_config.mark_pool != -1) {
@@ -234,6 +243,7 @@ int main(int argc, char **argv)
 {
 	signal(SIGHUP, hup);
 	signal(SIGINT, hup);
+	signal(SIGUSR1, do_nothing);  // Used in internal signalling.
 	signal(SIGPIPE, SIG_IGN);
 	
 	// Parse options.
@@ -337,8 +347,21 @@ start:
 
 		// Deserialize the streams.
 		for (int i = 0; i < loaded_state.streams_size(); ++i) {
-			servers->add_stream_from_serialized(loaded_state.streams(i));
-			deserialized_stream_ids.insert(loaded_state.streams(i).stream_id());
+			const StreamProto &stream = loaded_state.streams(i);
+
+			vector<int> data_fds;
+			for (int j = 0; j < stream.data_fds_size(); ++j) {
+				data_fds.push_back(stream.data_fds(j));
+			}
+
+			// Older versions stored the data once in the protobuf instead of
+			// sending around file descriptors.
+			if (data_fds.empty() && stream.has_data()) {
+				data_fds.push_back(make_tempfile(stream.data()));
+			}
+
+			servers->add_stream_from_serialized(stream, data_fds);
+			deserialized_stream_ids.insert(stream.stream_id());
 		}
 
 		// Deserialize the inputs. Note that we don't actually add them to any stream yet.
@@ -381,6 +404,7 @@ start:
 		if (input_it->second.refcount == 0) {
 			log(WARNING, "Input '%s' no longer in use, closing.",
 			    input_it->first.c_str());
+			input_it->second.input->close_socket();
 			delete input_it->second.input;
 			inputs.erase(input_it++);
 		} else {