socket stuff

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

client/Makefile

@@ -3,7 +3,7 @@ CC      = /usr/bin/g++
 CFLAGS  = -std=gnu++0x -Wall -Wno-unused-function -g -D_REENTRANT -DVERSION=\"$(VERSION)\" -I..
 LDFLAGS = -lrt -lpthread -lboost_system -lboost_program_options -lasound -logg -lvorbis -lvorbisenc
 
-OBJ = snapClient.o stream.o player.o socketConnection.o oggDecoder.o pcmDecoder.o controller.o ../common/pcmChunk.o ../common/log.o ../common/sampleFormat.o
+OBJ = snapClient.o stream.o player.o ../common/socketConnection.o oggDecoder.o pcmDecoder.o controller.o ../common/pcmChunk.o ../common/log.o ../common/sampleFormat.o
 BIN = snapclient
 
 all:	client

client/controller.cpp

@@ -59,6 +59,9 @@ void Controller::start(std::string& _ip, size_t _port, int _bufferMs)
 	connection = new ClientConnection(this, _ip, _port);
 	connection->start();
 
+//	controlConnection = new ClientConnection(this, _ip, _port + 1);
+//	controlConnection->start();
+
 	controllerThread = new thread(&Controller::worker, this);
 }
 
@@ -75,7 +78,9 @@ void Controller::worker()
 	active_ = true;	
 
 	while (sampleFormat == NULL)
+	{
 		usleep(10000);
+	}
 	stream = new Stream(SampleFormat(*sampleFormat));
 	stream->setBufferLen(bufferMs);
 
@@ -84,7 +89,9 @@ void Controller::worker()
 
 	while (active_)
 	{
-		usleep(10000);
+		usleep(100000);
+//		BaseMessage msg;
+//		controlConnection->send(&msg);
 	}
 }
 

client/controller.h

@@ -4,8 +4,9 @@
 #include <thread>
 #include <atomic>
 #include "common/message.h"
+#include "common/socketConnection.h"
 #include "decoder.h"
-#include "socketConnection.h"
+#include "stream.h"
 
 
 class Controller : public MessageReceiver
@@ -21,6 +22,7 @@ private:
 	std::atomic<bool> active_;
 	std::thread* controllerThread;
 	ClientConnection* connection;
+	ClientConnection* controlConnection;
 	SampleFormat* sampleFormat;
 	Decoder* decoder;
 	Stream* stream;

common/message.h

@@ -27,23 +27,23 @@ struct membuf : public std::basic_streambuf<char>
 
 enum message_type
 {
-	header = 0,
+	base = 0,
-	payload = 1,
+	header = 1,
-	sampleformat = 2
+	payload = 2,
+	sampleformat = 3
 };
 
 
 
 struct BaseMessage
 {
-	BaseMessage()
+	BaseMessage() : type(base)
 	{
 	}
 
-	BaseMessage(message_type type_)
+	BaseMessage(message_type type_) : type(type_) 
 	{
-		type = type_;
+	}
-	};
 
 	virtual ~BaseMessage()
 	{

common/socketConnection.cpp

@@ -2,7 +2,7 @@
 #include <boost/lexical_cast.hpp>
 #include <iostream>
 #include <mutex>
-#include "common/log.h"
+#include "log.h"
 
 
 #define PCM_DEVICE "default"
@@ -23,7 +23,7 @@ SocketConnection::~SocketConnection()
 
 void SocketConnection::socketRead(void* _to, size_t _bytes)
 {
-	std::unique_lock<std::mutex> mlock(mutex_);
+//	std::unique_lock<std::mutex> mlock(mutex_);
 	size_t toRead = _bytes;
 	size_t len = 0;
 	do 
@@ -52,12 +52,9 @@ void SocketConnection::send(BaseMessage* message)
 	std::unique_lock<std::mutex> mlock(mutex_);
 	boost::asio::streambuf streambuf;
 	std::ostream stream(&streambuf);
-	for (;;)
-	{
 	message->serialize(stream);
 	boost::asio::write(*socket.get(), streambuf);
 }
-}
 
 
 void SocketConnection::getNextMessage()
@@ -95,6 +92,8 @@ void ClientConnection::worker()
 	{
 		try
 		{	
+			{
+				std::unique_lock<std::mutex> mlock(mutex_);
 				tcp::resolver resolver(io_service);
 				tcp::resolver::query query(tcp::v4(), ip, boost::lexical_cast<string>(port));
 				iterator = resolver.resolve(query);
@@ -106,7 +105,7 @@ void ClientConnection::worker()
 				setsockopt(socket->native(), SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv));
 				socket->connect(*iterator);
 				std::clog << kLogNotice << "connected\n";// to " << ip << ":" << port << std::endl;
-
+			}
 			while(active_)
 			{
 				getNextMessage();

common/socketConnection.h

@@ -4,9 +4,9 @@
 #include <string>
 #include <thread>
 #include <atomic>
+#include <mutex>
 #include <boost/asio.hpp>
-#include "stream.h"
+#include "message.h"
-#include "common/message.h"
 
 
 using boost::asio::ip::tcp;

server/Makefile

@@ -3,7 +3,7 @@ CC      = /usr/bin/g++
 CFLAGS  = -std=gnu++0x -Wall -Wno-unused-function -D_REENTRANT -DVERSION=\"$(VERSION)\" -I..
 LDFLAGS = -lrt -lpthread -lboost_system -lboost_program_options -lvorbis -lvorbisenc -logg
 
-OBJ = snapServer.o streamServer.o controlServer.o pcmEncoder.o oggEncoder.o ../common/pcmChunk.o ../common/sampleFormat.o
+OBJ = snapServer.o streamServer.o controlServer.o pcmEncoder.o oggEncoder.o ../common/log.o ../common/socketConnection.o ../common/pcmChunk.o ../common/sampleFormat.o
 BIN = snapserver
 
 all:	server

server/controlServer.cpp

@@ -1,53 +1,5 @@
 #include "controlServer.h"
-
+#include <iostream>
-
-ControlSession::ControlSession(socket_ptr sock) : active_(false), socket_(sock)
-{
-}
-
-
-void ControlSession::sender()
-{
-	try
-	{
-		boost::asio::streambuf streambuf;
-		std::ostream stream(&streambuf);
-		for (;;)
-		{
-			shared_ptr<BaseMessage> message(messages.pop());
-			message->serialize(stream);
-			boost::asio::write(*socket_, streambuf);
-		}
-	}
-	catch (std::exception& e)
-	{
-		std::cerr << "Exception in thread: " << e.what() << "\n";
-		active_ = false;
-	}
-}
-
-void ControlSession::start()
-{
-	active_ = true;
-	senderThread = new thread(&ControlSession::sender, this);
-//		readerThread.join();
-}
-
-void ControlSession::send(BaseMessage* message)
-{
-	boost::asio::streambuf streambuf;
-	std::ostream stream(&streambuf);
-	message->serialize(stream);
-	boost::asio::write(*socket_, streambuf);
-}
-
-
-bool ControlSession::isActive() const
-{
-	return active_;
-}
-
-
 
 
 ControlServer::ControlServer(unsigned short port) : port_(port), headerChunk(NULL)
@@ -55,6 +7,12 @@ ControlServer::ControlServer(unsigned short port) : port_(port), headerChunk(NUL
 }
 
 
+void ControlServer::onMessageReceived(SocketConnection* connection, const BaseMessage& baseMessage, char* buffer)
+{
+	cout << "onMessageReceived: " << baseMessage.type << ", " << baseMessage.size << "\n";
+}
+
+
 void ControlServer::acceptor()
 {
 	tcp::acceptor a(io_service_, tcp::endpoint(tcp::v4(), port_));
@@ -63,8 +21,8 @@ void ControlServer::acceptor()
 		socket_ptr sock(new tcp::socket(io_service_));
 		a.accept(*sock);
 		cout << "New connection: " << sock->remote_endpoint().address().to_string() << "\n";
-		ControlSession* session = new ControlSession(sock);
+		ServerConnection* session = new ServerConnection(this, sock);
-		sessions.insert(shared_ptr<ControlSession>(session));
+		sessions.insert(shared_ptr<ServerConnection>(session));
 		session->start();
 	}
 }

server/controlServer.h

@@ -12,6 +12,7 @@
 #include "common/message.h"
 #include "common/headerMessage.h"
 #include "common/sampleFormat.h"
+#include "common/socketConnection.h"
 
 
 using boost::asio::ip::tcp;
@@ -19,37 +20,18 @@ typedef std::shared_ptr<tcp::socket> socket_ptr;
 using namespace std;
 
 
-
+class ControlServer : public MessageReceiver
-class ControlSession
-{
-public:
-	ControlSession(socket_ptr sock);
-
-	void start();
-	void send(BaseMessage* message);
-	bool isActive() const;
-
-private:
-	void sender();
-	bool active_;
-	socket_ptr socket_;
-	thread* senderThread;
-	Queue<shared_ptr<BaseMessage>> messages;
-};
-
-
-
-class ControlServer
 {
 public:
 	ControlServer(unsigned short port);
 
 	void start();
 	void stop();
+	virtual void onMessageReceived(SocketConnection* connection, const BaseMessage& baseMessage, char* buffer);
 
 private:
 	void acceptor();
-	set<shared_ptr<ControlSession>> sessions;
+	set<shared_ptr<ServerConnection>> sessions;
 	boost::asio::io_service io_service_;
 	unsigned short port_;
 	shared_ptr<HeaderMessage> headerChunk;

server/snapServer.cpp

@@ -64,6 +64,9 @@ int main(int argc, char* argv[])
 		StreamServer* server = new StreamServer(port);
 		server->start();
 
+		ControlServer* controlServer = new ControlServer(port + 1);
+		controlServer->start();
+
 		timeval tvChunk;
 		gettimeofday(&tvChunk, NULL);
 		long nextTick = getTickCount();