server Settings

git-svn-id: svn://elaine/murooma/trunk@255 d8a302eb-03bc-478d-80e4-98257eca68ef

client/controller.cpp

@@ -5,12 +5,12 @@
 #include "oggDecoder.h"
 #include "pcmDecoder.h"
 #include "player.h"
-
+#include "common/serverSettings.h"
 
 using namespace std;
 
 
-Controller::Controller() : MessageReceiver(), active_(false), sampleFormat(NULL)
+Controller::Controller() : MessageReceiver(), active_(false), streamClient(NULL), sampleFormat(NULL)
 {
 	decoder = new OggDecoder();
 }
@@ -18,6 +18,7 @@ Controller::Controller() : MessageReceiver(), active_(false), sampleFormat(NULL)
 
 void Controller::onMessageReceived(SocketConnection* connection, const BaseMessage& baseMessage, char* buffer)
 {
+//cout << "onMessageReceived: " << baseMessage.type << ", size: " << baseMessage.size << "\n"; 
 	if (baseMessage.type == message_type::payload)
 	{
 		if ((stream != NULL) && (decoder != NULL))
@@ -49,18 +50,23 @@ void Controller::onMessageReceived(SocketConnection* connection, const BaseMessa
 		sampleFormat->deserialize(baseMessage, buffer);
 		cout << "SampleFormat rate: " << sampleFormat->rate << ", bits: " << sampleFormat->bits << ", channels: " << sampleFormat->channels << "\n";
 	}
+	else if (baseMessage.type == message_type::serversettings)
+	{
+		ServerSettings* serverSettings = new ServerSettings();
+		serverSettings->deserialize(baseMessage, buffer);
+		cout << "ServerSettings port: " << serverSettings->port << "\n";
+		streamClient = new StreamClient(this, ip, serverSettings->port);
+	}
 }
 
 
-void Controller::start(std::string& _ip, size_t _port, int _bufferMs)
+void Controller::start(const std::string& _ip, size_t _port, int _bufferMs)
 {
 	bufferMs = _bufferMs;
+	ip = _ip;
 
-	connection = new ClientConnection(this, _ip, _port);
+	controlConnection = new ClientConnection(this, ip, _port);
-	connection->start();
+	controlConnection->start();
-
-//	controlConnection = new ClientConnection(this, _ip, _port + 1);
-//	controlConnection->start();
 
 	controllerThread = new thread(&Controller::worker, this);
 }
@@ -77,10 +83,13 @@ void Controller::worker()
 //	Decoder* decoder;
 	active_ = true;	
 
-	while (sampleFormat == NULL)
+	while ((sampleFormat == NULL) && (streamClient == NULL))
 	{
 		usleep(10000);
 	}
+	
+	streamClient->start();
+
 	stream = new Stream(SampleFormat(*sampleFormat));
 	stream->setBufferLen(bufferMs);
 

client/controller.h

@@ -5,6 +5,7 @@
 #include <atomic>
 #include "common/message.h"
 #include "common/socketConnection.h"
+#include "streamClient.h"
 #include "decoder.h"
 #include "stream.h"
 
@@ -13,7 +14,7 @@ class Controller : public MessageReceiver
 {
 public:
 	Controller();
-	void start(std::string& _ip, size_t _port, int _bufferMs);
+	void start(const std::string& _ip, size_t _port, int _bufferMs);
 	void stop();
 	virtual void onMessageReceived(SocketConnection* connection, const BaseMessage& baseMessage, char* buffer);
 
@@ -21,12 +22,13 @@ private:
 	void worker();
 	std::atomic<bool> active_;
 	std::thread* controllerThread;
-	ClientConnection* connection;
+	StreamClient* streamClient;
 	ClientConnection* controlConnection;
 	SampleFormat* sampleFormat;
 	Decoder* decoder;
 	Stream* stream;
 	int bufferMs;
+	std::string ip;
 };
 
 

common/message.h

@@ -30,7 +30,8 @@ enum message_type
 	base = 0,
 	header = 1,
 	payload = 2,
-	sampleformat = 3
+	sampleformat = 3,
+	serversettings = 4
 };
 
 

common/serverSettings.h

@@ -0,0 +1,43 @@
+#ifndef SERVER_SETTINGS_H
+#define SERVER_SETTINGS_H
+
+#include "message.h"
+
+
+
+class ServerSettings : public BaseMessage
+{
+public:
+	ServerSettings(size_t _port = 0) : BaseMessage(message_type::serversettings), port(_port)
+	{
+	}
+
+	virtual ~ServerSettings()
+	{
+	}
+
+	virtual void read(std::istream& stream)
+	{
+		stream.read(reinterpret_cast<char *>(&port), sizeof(int32_t));
+	}
+
+	virtual uint32_t getSize()
+	{
+		return sizeof(int32_t);
+	}
+
+	int32_t port;
+
+protected:
+	virtual void doserialize(std::ostream& stream)
+	{
+		stream.write(reinterpret_cast<char *>(&port), sizeof(int32_t));
+	}
+};
+
+
+
+
+#endif
+
+

common/socketConnection.cpp

@@ -50,8 +50,10 @@ void SocketConnection::stop()
 void SocketConnection::send(BaseMessage* message)
 {
 	std::unique_lock<std::mutex> mlock(mutex_);
+//cout << "send: " << message->type << ", size: " << message->getSize() << "\n";
 	if (!connected())
 		return;
+//cout << "send: " << message->type << ", size: " << message->getSize() << "\n";
 	boost::asio::streambuf streambuf;
 	std::ostream stream(&streambuf);
 	message->serialize(stream);
@@ -61,12 +63,13 @@ void SocketConnection::send(BaseMessage* message)
 
 void SocketConnection::getNextMessage()
 {
+//cout << "getNextMessage\n";
 	BaseMessage baseMessage;
 	size_t baseMsgSize = baseMessage.getSize();
 	vector<char> buffer(baseMsgSize);
 	socketRead(&buffer[0], baseMsgSize);
 	baseMessage.deserialize(&buffer[0]);
-//cout << "type: " << baseMessage.type << ", size: " << baseMessage.size << "\n";
+//cout << "getNextMessage: " << baseMessage.type << ", size: " << baseMessage.size << "\n";
 	if (baseMessage.size > buffer.size())
 		buffer.resize(baseMessage.size);
 	socketRead(&buffer[0], baseMessage.size);
@@ -102,7 +105,7 @@ void ClientConnection::worker()
 		{	
 			{
 //				std::unique_lock<std::mutex> mlock(mutex_);
-cout << "connecting\n";
+//cout << "connecting\n";
 				socket.reset(new tcp::socket(io_service));
 				struct timeval tv;
 				tv.tv_sec  = 5; 
@@ -110,7 +113,7 @@ cout << "connecting\n";
 				setsockopt(socket->native(), SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv));
 				socket->connect(*iterator);
 				connected_ = true;
-cout << "connected\n";
+//cout << "connected\n";
 				std::clog << kLogNotice << "connected\n";// to " << ip << ":" << port << std::endl;
 			}
 			while(active_)

common/socketConnection.h

@@ -38,7 +38,8 @@ public:
 
 	virtual bool connected() 
 	{
-		return (connected_ && socket);
+		return (socket != 0);
+//		return (connected_ && socket);
 	}
 
 protected:

server/controlServer.cpp

@@ -2,7 +2,7 @@
 #include <iostream>
 
 
-ControlServer::ControlServer(unsigned short port) : port_(port), headerChunk(NULL)
+ControlServer::ControlServer(unsigned short port) : port_(port), headerChunk(NULL), sampleFormat(NULL)
 {
 }
 
@@ -20,10 +20,13 @@ void ControlServer::acceptor()
 	{
 		socket_ptr sock(new tcp::socket(io_service_));
 		a.accept(*sock);
-		cout << "New connection: " << sock->remote_endpoint().address().to_string() << "\n";
+		cout << "ControlServer::NewConnection: " << sock->remote_endpoint().address().to_string() << "\n";
 		ServerConnection* session = new ServerConnection(this, sock);
 		sessions.insert(shared_ptr<ServerConnection>(session));
 		session->start();
+		session->send(serverSettings);
+		session->send(sampleFormat);
+		session->send(headerChunk);
 	}
 }
 
@@ -40,4 +43,28 @@ void ControlServer::stop()
 }
 
 
+void ControlServer::setHeader(HeaderMessage* header)
+{
+	if (header)
+		headerChunk = header;
+}
+
+
+void ControlServer::setFormat(SampleFormat* format)
+{
+	if (format)
+		sampleFormat = format;
+}
+
+
+
+void ControlServer::setServerSettings(ServerSettings* settings)
+{
+	if (settings)
+		serverSettings = settings;
+}
+
+
+
+
 

server/controlServer.h

@@ -12,6 +12,7 @@
 #include "common/message.h"
 #include "common/headerMessage.h"
 #include "common/sampleFormat.h"
+#include "common/serverSettings.h"
 #include "common/socketConnection.h"
 
 
@@ -28,14 +29,18 @@ public:
 	void start();
 	void stop();
 	virtual void onMessageReceived(SocketConnection* connection, const BaseMessage& baseMessage, char* buffer);
+	void setHeader(HeaderMessage* header);
+	void setFormat(SampleFormat* format);
+	void setServerSettings(ServerSettings* settings);
 
 private:
 	void acceptor();
 	set<shared_ptr<ServerConnection>> sessions;
 	boost::asio::io_service io_service_;
 	unsigned short port_;
-	shared_ptr<HeaderMessage> headerChunk;
+	HeaderMessage* headerChunk;
-	shared_ptr<SampleFormat> sampleFormat;
+	SampleFormat* sampleFormat;
+	ServerSettings* serverSettings;
 	thread* acceptThread;
 };
 

server/snapServer.cpp

@@ -61,11 +61,6 @@ int main(int argc, char* argv[])
 		openlog ("firstdaemon", LOG_PID, LOG_DAEMON);
 
 		using namespace std; // For atoi.
-		StreamServer* server = new StreamServer(port);
-		server->start();
-
-		ControlServer* controlServer = new ControlServer(port + 1);
-		controlServer->start();
 
 		timeval tvChunk;
 		gettimeofday(&tvChunk, NULL);
@@ -86,9 +81,15 @@ size_t duration = 50;
 			return 1;
 		}
 
-		server->setFormat(format);
+		std::auto_ptr<ServerSettings> serverSettings(new ServerSettings(port + 1));
-		shared_ptr<HeaderMessage> header(encoder->getHeader());
+		ControlServer* controlServer = new ControlServer(port);
-		server->setHeader(header);
+		controlServer->setServerSettings(serverSettings.get());
+		controlServer->setFormat(&format);
+		controlServer->setHeader(encoder->getHeader());
+		controlServer->start();
+
+		StreamServer* server = new StreamServer(port + 1);
+		server->start();
 
         while (!g_terminated)
         {

server/streamServer.cpp

@@ -57,32 +57,17 @@ void StreamServer::acceptor()
 		a.accept(*sock);
 		cout << "New connection: " << sock->remote_endpoint().address().to_string() << "\n";
 		StreamSession* session = new StreamSession(sock);
-		session->send(sampleFormat);
-		session->send(headerChunk);
-		session->start();
 		sessions.insert(shared_ptr<StreamSession>(session));
+		session->start();
 	}
 }
 
 
-void StreamServer::setHeader(shared_ptr<HeaderMessage> header)
-{
-	if (header)
-		headerChunk = header;
-}
-
-
-void StreamServer::setFormat(SampleFormat& format)
-{
-	sampleFormat = shared_ptr<SampleFormat>(new SampleFormat(format));
-}
-
-
 void StreamServer::send(shared_ptr<BaseMessage> message)
 {
 	for (std::set<shared_ptr<StreamSession>>::iterator it = sessions.begin(); it != sessions.end(); ) 
 	{
-		if (!(*it)->isActive())
+		if (!(*it)->active())
 		{
 			cout << "Session inactive. Removing\n";
 	        sessions.erase(it++);

server/streamServer.h

@@ -39,9 +39,6 @@ class StreamServer
 {
 public:
 	StreamServer(unsigned short port);
-
-	void setHeader(shared_ptr<HeaderMessage> header);
-	void setFormat(SampleFormat& format);
 	void send(shared_ptr<BaseMessage> message);
 
 	void start();