+ string stream_id = stream_config.stream_id;
+ string src = stream_config.src;
+
+ Input *input = NULL;
+ map<string, Input *>::iterator deserialized_input_it =
+ deserialized_inputs->find(stream_id);
+ if (deserialized_input_it != deserialized_inputs->end()) {
+ input = deserialized_input_it->second;
+ if (input->get_url() != src) {
+ fprintf(stderr, "INFO: Stream '%s' has changed URL from '%s' to '%s', restarting input.\n",
+ stream_id.c_str(), input->get_url().c_str(), src.c_str());
+ input->close_socket();
+ delete input;
+ input = NULL;
+ }
+ deserialized_inputs->erase(deserialized_input_it);
+ }
+ if (input == NULL) {
+ input = create_input(stream_id, src);
+ if (input == NULL) {
+ fprintf(stderr, "ERROR: did not understand URL '%s', clients will not get any data.\n",
+ src.c_str());
+ continue;
+ }
+ }
+ input->run();
+ inputs.push_back(input);
+ }
+ return inputs;
+}
+
+void create_streams(const Config &config,
+ const set<string> &deserialized_stream_ids,
+ map<string, Input *> *deserialized_inputs)
+{
+ vector<MarkPool *> mark_pools; // FIXME: leak
+ for (unsigned i = 0; i < config.mark_pools.size(); ++i) {
+ const MarkPoolConfig &mp_config = config.mark_pools[i];
+ mark_pools.push_back(new MarkPool(mp_config.from, mp_config.to));
+ }
+
+ set<string> expecting_stream_ids = deserialized_stream_ids;
+ 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);
+ }
+ expecting_stream_ids.erase(stream_config.stream_id);
+
+ if (stream_config.mark_pool != -1) {
+ servers->set_mark_pool(stream_config.stream_id,
+ mark_pools[stream_config.mark_pool]);
+ }
+ }
+
+ // Warn about any servers we've lost.
+ // TODO: Make an option (delete=yes?) to actually shut down streams.
+ for (set<string>::const_iterator stream_it = expecting_stream_ids.begin();
+ stream_it != expecting_stream_ids.end();
+ ++stream_it) {
+ string stream_id = *stream_it;
+ fprintf(stderr, "WARNING: stream '%s' disappeared from the configuration file.\n",
+ stream_id.c_str());
+ fprintf(stderr, " It will not be deleted, but clients will not get any new inputs.\n");
+ if (deserialized_inputs->count(stream_id) != 0) {
+ delete (*deserialized_inputs)[stream_id];
+ deserialized_inputs->erase(stream_id);
+ }
+ }