Support Metacube _output_. Required splitting HTTP headers from stream headers, which...

diff --git a/config.cpp b/config.cpp
index fad29a5361b7627eb442e0037c2595abcb2a55a7..c5547bda4f5d62ef155ffa4cf1381819658c90f4 100644 (file)
--- a/config.cpp
+++ b/config.cpp
@@ -207,6 +207,18 @@ bool parse_stream(const ConfigLine &line, Config *config)
                stream.backlog_size = atoi(backlog_it->second.c_str());
        }
 
                stream.backlog_size = atoi(backlog_it->second.c_str());
        }
 

+       // Parse encoding.
+       map<string, string>::const_iterator encoding_parm_it = line.parameters.find("encoding");
+       if (encoding_parm_it == line.parameters.end() ||
+           encoding_parm_it->second == "raw") {
+               stream.encoding = StreamConfig::STREAM_ENCODING_RAW;
+       } else if (encoding_parm_it->second == "metacube") {
+               stream.encoding = StreamConfig::STREAM_ENCODING_METACUBE;
+       } else {
+               log(ERROR, "Parameter 'encoding' must be either 'raw' (default) or 'metacube'");
+               return false;
+       }
+

        // Parse marks, if so desired.
        map<string, string>::const_iterator mark_parm_it = line.parameters.find("mark");
        if (mark_parm_it == line.parameters.end()) {
        // Parse marks, if so desired.
        map<string, string>::const_iterator mark_parm_it = line.parameters.find("mark");
        if (mark_parm_it == line.parameters.end()) {

diff --git a/config.h b/config.h
index 47dabe9d90a18cbb2b845cdbea65b9121ddc271f..5395d5affbb4e47e42d8fa1099864569973445fb 100644 (file)
--- a/config.h
+++ b/config.h
@@ -16,6 +16,7 @@ struct StreamConfig {
        std::string src;  // Can be empty.
        size_t backlog_size;
        int mark_pool;  // -1 for none.
        std::string src;  // Can be empty.
        size_t backlog_size;
        int mark_pool;  // -1 for none.

+       enum { STREAM_ENCODING_RAW = 0, STREAM_ENCODING_METACUBE } encoding;

 };
 
 struct AcceptorConfig {
 };
 
 struct AcceptorConfig {

diff --git a/cubemap.config.sample b/cubemap.config.sample
index 8504d2ad551784d79594ba153c009ff5e7f4013a..f3e2e0c689957715d17f9e33a1d4dbceeb0055a3 100644 (file)
--- a/cubemap.config.sample
+++ b/cubemap.config.sample
@@ -26,4 +26,5 @@ error_log type=console
 # now the streams!
 #
 stream /test.flv src=http://gruessi.zrh.sesse.net:4013/test.flv mark=1000-5000
 # now the streams!
 #
 stream /test.flv src=http://gruessi.zrh.sesse.net:4013/test.flv mark=1000-5000

+stream /test.flv.metacube src=http://gruessi.zrh.sesse.net:4013/test.flv encoding=metacube

 stream /udp.ts src=udp://@:1234 backlog_size=1048576
 stream /udp.ts src=udp://@:1234 backlog_size=1048576

diff --git a/httpinput.cpp b/httpinput.cpp
index 51b4f412582bb94d512b074249d5375f94804bfb..457c2160b241709b042ea473ec3a0c8803f3581e 100644 (file)
--- a/httpinput.cpp
+++ b/httpinput.cpp
@@ -50,6 +50,13 @@ HTTPInput::HTTPInput(const InputProto &serialized)
 
        string protocol;
        parse_url(url, &protocol, &host, &port, &path);  // Don't care if it fails.
 
        string protocol;
        parse_url(url, &protocol, &host, &port, &path);  // Don't care if it fails.

+
+       // Older versions stored the extra \r\n in the HTTP header.
+       // Strip it if we find it.
+       if (http_header.size() >= 4 &&
+           memcmp(http_header.data() + http_header.size() - 4, "\r\n\r\n", 4) == 0) {
+               http_header.resize(http_header.size() - 2);
+       }

 }
 
 void HTTPInput::close_socket()
 }
 
 void HTTPInput::close_socket()


@@ -200,10 +207,9 @@ bool HTTPInput::parse_response(const std::string &request)
             ++it) {
                http_header.append(it->first + ": " + it->second + "\r\n");
        }
             ++it) {
                http_header.append(it->first + ": " + it->second + "\r\n");
        }

-       http_header.append("\r\n");

 
        for (size_t i = 0; i < stream_ids.size(); ++i) {
 
        for (size_t i = 0; i < stream_ids.size(); ++i) {

-               servers->set_header(stream_ids[i], http_header);
+               servers->set_header(stream_ids[i], http_header, "");

        }
 
        return true;
        }
 
        return true;


@@ -237,7 +243,7 @@ void HTTPInput::do_work()
                        response.clear();
                        pending_data.clear();
                        for (size_t i = 0; i < stream_ids.size(); ++i) {
                        response.clear();
                        pending_data.clear();
                        for (size_t i = 0; i < stream_ids.size(); ++i) {

-                               servers->set_header(stream_ids[i], "");
+                               servers->set_header(stream_ids[i], "", "");

                        }
 
                        {
                        }
 
                        {


@@ -444,7 +450,7 @@ void HTTPInput::process_data(char *ptr, size_t bytes)
                if (flags & METACUBE_FLAGS_HEADER) {
                        string header(inner_data, inner_data + size);
                        for (size_t i = 0; i < stream_ids.size(); ++i) {
                if (flags & METACUBE_FLAGS_HEADER) {
                        string header(inner_data, inner_data + size);
                        for (size_t i = 0; i < stream_ids.size(); ++i) {

-                               servers->set_header(stream_ids[i], http_header + header);
+                               servers->set_header(stream_ids[i], http_header, header);

                        }
                } else { 
                        for (size_t i = 0; i < stream_ids.size(); ++i) {
                        }
                } else { 
                        for (size_t i = 0; i < stream_ids.size(); ++i) {

diff --git a/main.cpp b/main.cpp
index fa63549bd7f8a00b1ce05d55ad09c5f594507cfa..b45713db99ccd307442165f69150fa6492c3c9ea 100644 (file)
--- a/main.cpp
+++ b/main.cpp
@@ -143,9 +143,13 @@ 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) {
        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);
                } 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);
 
                }
                expecting_stream_ids.erase(stream_config.stream_id);
 

diff --git a/server.cpp b/server.cpp
index 8c3382d4e8c6506d4c6a99aa686c2bc5121fa436..e4ea647c64d05555505f3f18358f4e9b642d5370 100644 (file)
--- a/server.cpp
+++ b/server.cpp
@@ -1,3 +1,4 @@
+#include <arpa/inet.h>

 #include <assert.h>
 #include <errno.h>
 #include <pthread.h>
 #include <assert.h>
 #include <errno.h>
 #include <pthread.h>


@@ -17,6 +18,7 @@
 #include "accesslog.h"
 #include "log.h"
 #include "markpool.h"
 #include "accesslog.h"
 #include "log.h"
 #include "markpool.h"

+#include "metacube.h"

 #include "mutexlock.h"
 #include "parse.h"
 #include "server.h"
 #include "mutexlock.h"
 #include "parse.h"
 #include "server.h"


@@ -194,10 +196,10 @@ void Server::add_client_from_serialized(const ClientProto &client)
        }
 }
 
        }
 }
 

-void Server::add_stream(const string &stream_id, size_t backlog_size)
+void Server::add_stream(const string &stream_id, size_t backlog_size, Stream::Encoding encoding)

 {
        MutexLock lock(&mutex);
 {
        MutexLock lock(&mutex);

-       streams.insert(make_pair(stream_id, new Stream(stream_id, backlog_size)));
+       streams.insert(make_pair(stream_id, new Stream(stream_id, backlog_size, encoding)));

 }
 
 void Server::add_stream_from_serialized(const StreamProto &stream)
 }
 
 void Server::add_stream_from_serialized(const StreamProto &stream)


@@ -206,17 +208,25 @@ void Server::add_stream_from_serialized(const StreamProto &stream)
        streams.insert(make_pair(stream.stream_id(), new Stream(stream)));
 }
        
        streams.insert(make_pair(stream.stream_id(), new Stream(stream)));
 }
        

-void Server::set_backlog_size(const std::string &stream_id, size_t new_size)
+void Server::set_backlog_size(const string &stream_id, size_t new_size)

 {
        MutexLock lock(&mutex);
        assert(streams.count(stream_id) != 0);
        streams[stream_id]->set_backlog_size(new_size);
 }
        
 {
        MutexLock lock(&mutex);
        assert(streams.count(stream_id) != 0);
        streams[stream_id]->set_backlog_size(new_size);
 }
        

-void Server::set_header(const string &stream_id, const string &header)
+void Server::set_encoding(const string &stream_id, Stream::Encoding encoding)

 {
        MutexLock lock(&mutex);
 {
        MutexLock lock(&mutex);

-       find_stream(stream_id)->header = header;
+       assert(streams.count(stream_id) != 0);
+       streams[stream_id]->encoding = encoding;
+}
+       
+void Server::set_header(const string &stream_id, const string &http_header, const string &stream_header)
+{
+       MutexLock lock(&mutex);
+       find_stream(stream_id)->http_header = http_header;
+       find_stream(stream_id)->stream_header = stream_header;

 
        // If there are clients we haven't sent anything to yet, we should give
        // them the header, so push back into the SENDING_HEADER state.
 
        // If there are clients we haven't sent anything to yet, we should give
        // them the header, so push back into the SENDING_HEADER state.


@@ -231,7 +241,7 @@ void Server::set_header(const string &stream_id, const string &header)
        }
 }
        
        }
 }
        

-void Server::set_mark_pool(const std::string &stream_id, MarkPool *mark_pool)
+void Server::set_mark_pool(const string &stream_id, MarkPool *mark_pool)

 {
        MutexLock lock(&mutex);
        assert(clients.empty());
 {
        MutexLock lock(&mutex);
        assert(clients.empty());


@@ -454,7 +464,25 @@ int Server::parse_request(Client *client)
 
 void Server::construct_header(Client *client)
 {
 
 void Server::construct_header(Client *client)
 {

-       client->header_or_error = find_stream(client->stream_id)->header;
+       Stream *stream = find_stream(client->stream_id);
+       if (stream->encoding == Stream::STREAM_ENCODING_RAW) {
+               client->header_or_error = stream->http_header +
+                       "\r\n" +
+                       stream->stream_header;
+       } else if (stream->encoding == Stream::STREAM_ENCODING_METACUBE) {
+               metacube_block_header hdr;
+               memcpy(hdr.sync, METACUBE_SYNC, sizeof(hdr.sync));
+               hdr.size = htonl(stream->stream_header.size());
+               hdr.flags = htonl(METACUBE_FLAGS_HEADER);
+
+               client->header_or_error = stream->http_header +
+                       "Content-encoding: metacube\r\n" +
+                       "\r\n" +
+                       string(reinterpret_cast<char *>(&hdr), sizeof(hdr)) +
+                       stream->stream_header;
+       } else {
+               assert(false);
+       }

 
        // Switch states.
        client->state = Client::SENDING_HEADER;
 
        // Switch states.
        client->state = Client::SENDING_HEADER;

diff --git a/server.h b/server.h
index 2d227b6e1797ec331c7a6e966ec412d5f6c7a79b..523f5c55f376eed2ca9becf25a15c60a9172ba70 100644 (file)
--- a/server.h
+++ b/server.h
@@ -12,6 +12,7 @@
 #include <vector>
 
 #include "client.h"
 #include <vector>
 
 #include "client.h"

+#include "stream.h"

 #include "thread.h"
 
 class ClientProto;
 #include "thread.h"
 
 class ClientProto;


@@ -34,7 +35,9 @@ public:
        std::vector<ClientStats> get_client_stats() const;
 
        // Set header (both HTTP header and any stream headers) for the given stream.
        std::vector<ClientStats> get_client_stats() const;
 
        // Set header (both HTTP header and any stream headers) for the given stream.

-       void set_header(const std::string &stream_id, const std::string &header);
+       void set_header(const std::string &stream_id,
+                       const std::string &http_header,
+                       const std::string &stream_header);

 
        // Set that the given stream should use the given mark pool from now on.
        // NOTE: This should be set before any clients are connected!
 
        // Set that the given stream should use the given mark pool from now on.
        // NOTE: This should be set before any clients are connected!


@@ -51,9 +54,10 @@ public:
        // at the same time).
        CubemapStateProto serialize();
        void add_client_from_serialized(const ClientProto &client);
        // at the same time).
        CubemapStateProto serialize();
        void add_client_from_serialized(const ClientProto &client);

-       void add_stream(const std::string &stream_id, size_t bytes_received);
+       void add_stream(const std::string &stream_id, size_t bytes_received, Stream::Encoding encoding);

        void add_stream_from_serialized(const StreamProto &stream);
        void set_backlog_size(const std::string &stream_id, size_t new_size);
        void add_stream_from_serialized(const StreamProto &stream);
        void set_backlog_size(const std::string &stream_id, size_t new_size);

+       void set_encoding(const std::string &stream_id, Stream::Encoding encoding);

 
 private:
        // Mutex protecting queued_data only. Note that if you want to hold both this
 
 private:
        // Mutex protecting queued_data only. Note that if you want to hold both this

diff --git a/serverpool.cpp b/serverpool.cpp
index 352f7191f209a50630ede4c96a8aecb1ae33cd8a..60a913d56146ecf58f7f9b72b82af0bb7adeb1f5 100644 (file)
--- a/serverpool.cpp
+++ b/serverpool.cpp
@@ -48,10 +48,10 @@ void ServerPool::add_client_from_serialized(const ClientProto &client)
        servers[clients_added++ % num_servers].add_client_from_serialized(client);
 }
 
        servers[clients_added++ % num_servers].add_client_from_serialized(client);
 }
 

-void ServerPool::add_stream(const std::string &stream_id, size_t backlog_size)
+void ServerPool::add_stream(const string &stream_id, size_t backlog_size, Stream::Encoding encoding)

 {
        for (int i = 0; i < num_servers; ++i) {
 {
        for (int i = 0; i < num_servers; ++i) {

-               servers[i].add_stream(stream_id, backlog_size);
+               servers[i].add_stream(stream_id, backlog_size, encoding);

        }
 }
 
        }
 }
 


@@ -62,14 +62,14 @@ void ServerPool::add_stream_from_serialized(const StreamProto &stream)
        }
 }
 
        }
 }
 

-void ServerPool::set_header(const std::string &stream_id, const std::string &header)
+void ServerPool::set_header(const string &stream_id, const string &http_header, const string &stream_header)

 {
        for (int i = 0; i < num_servers; ++i) {
 {
        for (int i = 0; i < num_servers; ++i) {

-               servers[i].set_header(stream_id, header);
+               servers[i].set_header(stream_id, http_header, stream_header);

        }
 }
 
        }
 }
 

-void ServerPool::add_data(const std::string &stream_id, const char *data, size_t bytes)
+void ServerPool::add_data(const string &stream_id, const char *data, size_t bytes)

 {
        for (int i = 0; i < num_servers; ++i) {
                servers[i].add_data_deferred(stream_id, data, bytes);
 {
        for (int i = 0; i < num_servers; ++i) {
                servers[i].add_data_deferred(stream_id, data, bytes);


@@ -100,16 +100,23 @@ vector<ClientStats> ServerPool::get_client_stats() const
        return ret;
 }
        
        return ret;
 }
        

-void ServerPool::set_mark_pool(const std::string &stream_id, MarkPool *mark_pool)
+void ServerPool::set_mark_pool(const string &stream_id, MarkPool *mark_pool)

 {
        for (int i = 0; i < num_servers; ++i) {
                servers[i].set_mark_pool(stream_id, mark_pool);
        }       
 }
 
 {
        for (int i = 0; i < num_servers; ++i) {
                servers[i].set_mark_pool(stream_id, mark_pool);
        }       
 }
 

-void ServerPool::set_backlog_size(const std::string &stream_id, size_t new_size)
+void ServerPool::set_backlog_size(const string &stream_id, size_t new_size)

 {
        for (int i = 0; i < num_servers; ++i) {
                servers[i].set_backlog_size(stream_id, new_size);
        }       
 }
 {
        for (int i = 0; i < num_servers; ++i) {
                servers[i].set_backlog_size(stream_id, new_size);
        }       
 }

+
+void ServerPool::set_encoding(const string &stream_id, Stream::Encoding encoding)
+{
+       for (int i = 0; i < num_servers; ++i) {
+               servers[i].set_encoding(stream_id, encoding);
+       }       
+}

diff --git a/serverpool.h b/serverpool.h
index ba60d7c0d9b0705e01c372ed9f4d71d9db99e35d..84a201fb6fb2aea2748cad988e1254fe80e662ac 100644 (file)
--- a/serverpool.h
+++ b/serverpool.h
@@ -26,11 +26,13 @@ public:
        void add_client_from_serialized(const ClientProto &client);
 
        // Adds the given stream to all the servers.
        void add_client_from_serialized(const ClientProto &client);
 
        // Adds the given stream to all the servers.

-       void add_stream(const std::string &stream_id, size_t backlog_size);
+       void add_stream(const std::string &stream_id, size_t backlog_size, Stream::Encoding encoding);

        void add_stream_from_serialized(const StreamProto &stream);
 
        // Adds the given data to all the servers.
        void add_stream_from_serialized(const StreamProto &stream);
 
        // Adds the given data to all the servers.

-       void set_header(const std::string &stream_id, const std::string &header);
+       void set_header(const std::string &stream_id,
+                       const std::string &http_header,
+                       const std::string &stream_header);

        void add_data(const std::string &stream_id, const char *data, size_t bytes);
 
        // Connects the given stream to the given mark pool for all the servers.
        void add_data(const std::string &stream_id, const char *data, size_t bytes);
 
        // Connects the given stream to the given mark pool for all the servers.


@@ -39,6 +41,9 @@ public:
        // Changes the given stream's backlog size on all the servers.
        void set_backlog_size(const std::string &stream_id, size_t new_size);
 
        // Changes the given stream's backlog size on all the servers.
        void set_backlog_size(const std::string &stream_id, size_t new_size);
 

+       // Changes the given stream's encoding type on all the servers.
+       void set_encoding(const std::string &stream_id, Stream::Encoding encoding);
+

        // Starts all the servers.
        void run();
 
        // Starts all the servers.
        void run();
 

diff --git a/state.proto b/state.proto
index 87b028c9dae465f5202dca469c8702f7f8b24090..5d38a6bd06df19f66e09baad11b21ee00f83214c 100644 (file)
--- a/state.proto
+++ b/state.proto
@@ -16,11 +16,15 @@ message ClientProto {
 
 // Corresponds to struct Stream.
 message StreamProto {
 
 // Corresponds to struct Stream.
 message StreamProto {

-       optional bytes header = 1;
+       optional bytes http_header = 6;
+       optional bytes stream_header = 7;

        optional bytes data = 2;
        optional int64 backlog_size = 5 [default=1048576];
        optional int64 bytes_received = 3;
        optional string stream_id = 4;
        optional bytes data = 2;
        optional int64 backlog_size = 5 [default=1048576];
        optional int64 bytes_received = 3;
        optional string stream_id = 4;

+
+       // Older versions stored the HTTP and video headers together in this field.
+       optional bytes header = 1;

 };
 
 // Corresponds to class Input.
 };
 
 // Corresponds to class Input.

diff --git a/stream.cpp b/stream.cpp
index 2b9a69593a478605800fd21c9585063eebec8d0c..e61428a97318ec6be30e6a44964a79ac571a7f4a 100644 (file)
--- a/stream.cpp
+++ b/stream.cpp
@@ -1,3 +1,4 @@
+#include <arpa/inet.h>

 #include <errno.h>
 #include <stdio.h>
 #include <stdlib.h>
 #include <errno.h>
 #include <stdio.h>
 #include <stdlib.h>


@@ -7,13 +8,15 @@
 
 #include "state.pb.h"
 #include "log.h"
 
 #include "state.pb.h"
 #include "log.h"

+#include "metacube.h"

 #include "stream.h"
 #include "util.h"
 
 using namespace std;
 
 #include "stream.h"
 #include "util.h"
 
 using namespace std;
 

-Stream::Stream(const string &stream_id, size_t backlog_size)
+Stream::Stream(const string &stream_id, size_t backlog_size, Encoding encoding)

        : stream_id(stream_id),
        : stream_id(stream_id),

+         encoding(encoding),

          data_fd(make_tempfile("")),
           backlog_size(backlog_size),
          bytes_received(0),
          data_fd(make_tempfile("")),
           backlog_size(backlog_size),
          bytes_received(0),


@@ -39,7 +42,9 @@ Stream::~Stream()
 
 Stream::Stream(const StreamProto &serialized)
        : stream_id(serialized.stream_id()),
 
 Stream::Stream(const StreamProto &serialized)
        : stream_id(serialized.stream_id()),

-         header(serialized.header()),
+         http_header(serialized.http_header()),
+         stream_header(serialized.stream_header()),
+         encoding(Stream::STREAM_ENCODING_RAW),  // Will be changed later.

          data_fd(make_tempfile(serialized.data())),
          backlog_size(serialized.backlog_size()),
          bytes_received(serialized.bytes_received()),
          data_fd(make_tempfile(serialized.data())),
          backlog_size(serialized.backlog_size()),
          bytes_received(serialized.bytes_received()),


@@ -48,12 +53,26 @@ Stream::Stream(const StreamProto &serialized)
        if (data_fd == -1) {
                exit(1);
        }
        if (data_fd == -1) {
                exit(1);
        }

+
+       // Split old-style headers into HTTP and video headers.
+       if (!serialized.header().empty()) {
+               string header = serialized.header();
+               size_t split = header.find("\r\n\r\n");
+               if (split == string::npos) {
+                       http_header = header;
+                       stream_header = "";
+               } else {
+                       http_header = header.substr(0, split + 2);  // Split off the second \r\n.
+                       stream_header = header.substr(split, string::npos);
+               }
+       }

 }
 
 StreamProto Stream::serialize()
 {
        StreamProto serialized;
 }
 
 StreamProto Stream::serialize()
 {
        StreamProto serialized;

-       serialized.set_header(header);
+       serialized.set_http_header(http_header);
+       serialized.set_stream_header(stream_header);

        if (!read_tempfile(data_fd, serialized.mutable_data())) {  // Closes data_fd.
                exit(1);
        }
        if (!read_tempfile(data_fd, serialized.mutable_data())) {  // Closes data_fd.
                exit(1);
        }


@@ -105,6 +124,26 @@ void Stream::put_client_to_sleep(Client *client)
 }
 
 void Stream::add_data(const char *data, ssize_t bytes)
 }
 
 void Stream::add_data(const char *data, ssize_t bytes)

+{
+       if (encoding == Stream::STREAM_ENCODING_RAW) {
+               add_data_raw(data, bytes);
+       } else if (encoding == STREAM_ENCODING_METACUBE) {
+               metacube_block_header hdr;
+               memcpy(hdr.sync, METACUBE_SYNC, sizeof(hdr.sync));
+               hdr.size = htonl(bytes);
+               hdr.flags = htonl(0);
+
+               char *block = new char[bytes + sizeof(hdr)];
+               memcpy(block, &hdr, sizeof(hdr));
+               memcpy(block + sizeof(hdr), data, bytes);
+               add_data_raw(block, bytes + sizeof(hdr));
+               delete[] block;
+       } else {
+               assert(false);
+       }
+}
+
+void Stream::add_data_raw(const char *data, ssize_t bytes)

 {
        size_t pos = bytes_received % backlog_size;
        bytes_received += bytes;
 {
        size_t pos = bytes_received % backlog_size;
        bytes_received += bytes;

diff --git a/stream.h b/stream.h
index a4aa642b1b81cbb5a336cd138e3fe8d94122c6f7..22a162df0f753bbb70fa3844ba534e75d3bb666c 100644 (file)
--- a/stream.h
+++ b/stream.h
@@ -15,7 +15,10 @@ class StreamProto;
 struct Client;
 
 struct Stream {
 struct Client;
 
 struct Stream {

-       Stream(const std::string &stream_id, size_t backlog_size);
+       // Must be in sync with StreamConfig::Encoding.
+       enum Encoding { STREAM_ENCODING_RAW = 0, STREAM_ENCODING_METACUBE };
+
+       Stream(const std::string &stream_id, size_t backlog_size, Encoding encoding);

        ~Stream();
 
        // Serialization/deserialization.
        ~Stream();
 
        // Serialization/deserialization.


@@ -27,8 +30,15 @@ struct Stream {
 
        std::string stream_id;
 
 
        std::string stream_id;
 

-       // The HTTP response header, plus the video stream header (if any).
-       std::string header;
+       // The HTTP response header, without the trailing double newline.
+       std::string http_header;
+
+       // The video stream header (if any).
+       std::string stream_header;
+
+       // What encoding we apply to the outgoing data (usually raw, but can also
+       // be Metacube, for reflecting to another Cubemap instance).
+       Encoding encoding;

 
        // The stream data itself, stored in a circular buffer.
        //
 
        // The stream data itself, stored in a circular buffer.
        //


@@ -73,6 +83,8 @@ struct Stream {
 
 private:
        Stream(const Stream& other);
 
 private:
        Stream(const Stream& other);

+
+       void add_data_raw(const char *data, ssize_t bytes);

 };
 
 #endif  // !defined(_STREAM_H)
 };
 
 #endif  // !defined(_STREAM_H)

diff --git a/udpinput.cpp b/udpinput.cpp
index bd83381b563dc9b897b70f8d89b66a39a3da0ef1..ae4f288e34fc16da66470f656f6ef369f3322be3 100644 (file)
--- a/udpinput.cpp
+++ b/udpinput.cpp
@@ -70,10 +70,9 @@ void UDPInput::construct_header()
                "HTTP/1.0 200 OK\r\n"
                "Content-type: application/octet-stream\r\n"
                "Cache-control: no-cache\r\n"
                "HTTP/1.0 200 OK\r\n"
                "Content-type: application/octet-stream\r\n"
                "Cache-control: no-cache\r\n"

-               "Server: " SERVER_IDENTIFICATION "\r\n"
-               "\r\n";
+               "Server: " SERVER_IDENTIFICATION "\r\n";

        for (size_t i = 0; i < stream_ids.size(); ++i) {
        for (size_t i = 0; i < stream_ids.size(); ++i) {

-               servers->set_header(stream_ids[i], header);
+               servers->set_header(stream_ids[i], header, "");

        }
 }
 
        }
 }