socket

git-svn-id: svn://elaine/murooma/trunk@250 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 serverConnection.o oggDecoder.o pcmDecoder.o controller.o ../common/pcmChunk.o ../common/log.o ../common/sampleFormat.o
+OBJ = snapClient.o stream.o player.o socketConnection.o oggDecoder.o pcmDecoder.o controller.o ../common/pcmChunk.o ../common/log.o ../common/sampleFormat.o
 BIN = snapclient
 
 all:	client

client/controller.cpp

@@ -16,7 +16,7 @@ Controller::Controller() : MessageReceiver(), active_(false), sampleFormat(NULL)
 }
 
 
-void Controller::onMessageReceived(tcp::socket* socket, const BaseMessage& baseMessage, char* buffer)
+void Controller::onMessageReceived(SocketConnection* connection, const BaseMessage& baseMessage, char* buffer)
 {
 	if (baseMessage.type == message_type::payload)
 	{
@@ -56,8 +56,8 @@ void Controller::start(std::string& _ip, size_t _port, int _bufferMs)
 {
 	bufferMs = _bufferMs;
 
-	connection = new ServerConnection();
+	connection = new ClientConnection(this, _ip, _port);
-	connection->start(this, _ip, _port);
+	connection->start();
 
 	controllerThread = new thread(&Controller::worker, this);
 }

client/controller.h

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

client/serverConnection.cpp

@@ -1,98 +0,0 @@
-#include "serverConnection.h"
-#include <boost/lexical_cast.hpp>
-#include <iostream>
-#include "common/log.h"
-#include "common/message.h"
-#include "common/headerMessage.h"
-
-
-#define PCM_DEVICE "default"
-
-using namespace std;
-
-
-ServerConnection::ServerConnection() : active_(false)
-{
-}
-
-
-void ServerConnection::socketRead(tcp::socket* socket, void* to, size_t bytes)
-{
-	size_t toRead = bytes;
-	size_t len = 0;
-	do 
-	{
-		len += socket->read_some(boost::asio::buffer((char*)to + len, toRead));
-		toRead = bytes - len;
-	}
-	while (toRead > 0);
-}
-
-
-void ServerConnection::start(MessageReceiver* receiver, const std::string& ip, size_t port)
-{
-	messageReceiver = receiver;
-//	endpt.address(boost::asio::ip::address::from_string(ip));
-//	endpt.port((port));
-//    std::cout << "Endpoint IP:   " << endpt.address().to_string() << std::endl;
-//    std::cout << "Endpoint Port: " << endpt.port() << std::endl;
-	tcp::resolver resolver(io_service);
-	tcp::resolver::query query(tcp::v4(), ip, boost::lexical_cast<string>(port));
-	iterator = resolver.resolve(query);
-	receiverThread = new thread(&ServerConnection::worker, this);
-}
-
-
-void ServerConnection::stop()
-{
-	active_ = false;
-}
-
-
-void ServerConnection::getNextMessage(tcp::socket* socket)
-{
-	BaseMessage baseMessage;
-	size_t baseMsgSize = baseMessage.getSize();
-	vector<char> buffer(baseMsgSize);
-	socketRead(socket, &buffer[0], baseMsgSize);
-	baseMessage.deserialize(&buffer[0]);
-//cout << "type: " << baseMessage.type << ", size: " << baseMessage.size << "\n";
-	if (baseMessage.size > buffer.size())
-		buffer.resize(baseMessage.size);
-	socketRead(socket, &buffer[0], baseMessage.size);
-	
-	if (messageReceiver != NULL)
-		messageReceiver->onMessageReceived(socket, baseMessage, &buffer[0]);
-}
-
-
-void ServerConnection::worker() 
-{
-	active_ = true;
-	while (active_)
-	{
-		try
-		{
-			tcp::socket s(io_service);
-			s.connect(*iterator);
-			struct timeval tv;
-			tv.tv_sec  = 5; 
-			tv.tv_usec = 0;         
-			setsockopt(s.native(), SOL_SOCKET, SO_RCVTIMEO, &tv, sizeof(tv));
-			std::clog << kLogNotice << "connected\n";// to " << ip << ":" << port << std::endl;
-
-			while(active_)
-			{
-				getNextMessage(&s);
-			}
-		}
-		catch (const std::exception& e)
-		{
-			cout << kLogNotice << "Exception: " << e.what() << ", trying to reconnect" << std::endl;
-			usleep(500*1000);
-		}
-	}
-}
-
-
-

client/serverConnection.h

@@ -1,43 +0,0 @@
-#ifndef RECEIVER_H
-#define RECEIVER_H
-
-#include <string>
-#include <thread>
-#include <atomic>
-#include <boost/asio.hpp>
-#include "stream.h"
-
-
-using boost::asio::ip::tcp;
-
-
-class MessageReceiver
-{
-public:
-	virtual void onMessageReceived(tcp::socket* socket, const BaseMessage& baseMessage, char* buffer) = 0;
-};
-
-
-class ServerConnection
-{
-public:
-	ServerConnection();
-	void start(MessageReceiver* receiver, const std::string& ip, size_t port);
-	void stop();
-
-private:
-	void socketRead(tcp::socket* socket, void* to, size_t bytes);
-	void worker();
-
-//	boost::asio::ip::tcp::endpoint endpt;
-	MessageReceiver* messageReceiver;
-	void getNextMessage(tcp::socket* socket);
-	boost::asio::io_service io_service;
-	tcp::resolver::iterator iterator;
-	std::atomic<bool> active_;
-	std::thread* receiverThread;
-};
-
-
-#endif
-

client/socketConnection.cpp

@@ -0,0 +1,158 @@
+#include "socketConnection.h"
+#include <boost/lexical_cast.hpp>
+#include <iostream>
+#include <mutex>
+#include "common/log.h"
+
+
+#define PCM_DEVICE "default"
+
+using namespace std;
+
+
+SocketConnection::SocketConnection(MessageReceiver* _receiver) : active_(false), messageReceiver(_receiver)
+{
+}
+
+
+SocketConnection::~SocketConnection()
+{
+}
+
+
+
+void SocketConnection::socketRead(void* _to, size_t _bytes)
+{
+	std::unique_lock<std::mutex> mlock(mutex_);
+	size_t toRead = _bytes;
+	size_t len = 0;
+	do 
+	{
+		len += socket->read_some(boost::asio::buffer((char*)_to + len, toRead));
+		toRead = _bytes - len;
+	}
+	while (toRead > 0);
+}
+
+
+void SocketConnection::start()
+{
+	receiverThread = new thread(&SocketConnection::worker, this);
+}
+
+
+void SocketConnection::stop()
+{
+	active_ = false;
+}
+
+
+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()
+{
+	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";
+	if (baseMessage.size > buffer.size())
+		buffer.resize(baseMessage.size);
+	socketRead(&buffer[0], baseMessage.size);
+	
+	if (messageReceiver != NULL)
+		messageReceiver->onMessageReceived(this, baseMessage, &buffer[0]);
+}
+
+
+
+
+ClientConnection::ClientConnection(MessageReceiver* _receiver, const std::string& _ip, size_t _port) : SocketConnection(_receiver), ip(_ip), port(_port)
+{
+//	endpt.address(boost::asio::ip::address::from_string(ip));
+//	endpt.port((port));
+//    std::cout << "Endpoint IP:   " << endpt.address().to_string() << std::endl;
+//    std::cout << "Endpoint Port: " << endpt.port() << std::endl;
+}
+
+
+void ClientConnection::worker() 
+{
+	active_ = true;
+	while (active_)
+	{
+		try
+		{
+			tcp::resolver resolver(io_service);
+			tcp::resolver::query query(tcp::v4(), ip, boost::lexical_cast<string>(port));
+			iterator = resolver.resolve(query);
+
+			socket.reset(new tcp::socket(io_service));
+			struct timeval tv;
+			tv.tv_sec  = 5; 
+			tv.tv_usec = 0;         
+			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();
+			}
+		}
+		catch (const std::exception& e)
+		{
+			cout << kLogNotice << "Exception: " << e.what() << ", trying to reconnect" << std::endl;
+			usleep(500*1000);
+		}
+	}
+}
+
+
+
+
+
+ServerConnection::ServerConnection(MessageReceiver* _receiver, std::shared_ptr<tcp::socket> _socket) : SocketConnection(_receiver)
+{
+	socket = _socket;
+}
+
+
+void ServerConnection::worker()
+{
+	active_ = true;
+	try
+	{
+		while (active_)
+		{
+			getNextMessage();
+		}
+	}
+	catch (const std::exception& e)
+	{
+		cout << kLogNotice << "Exception: " << e.what() << ", trying to reconnect" << std::endl;
+	}
+	active_ = false;
+}
+
+
+
+
+
+
+
+
+
+

client/socketConnection.h

@@ -0,0 +1,82 @@
+#ifndef SOCKET_CONNECTION_H
+#define SOCKET_CONNECTION_H
+
+#include <string>
+#include <thread>
+#include <atomic>
+#include <boost/asio.hpp>
+#include "stream.h"
+#include "common/message.h"
+
+
+using boost::asio::ip::tcp;
+
+
+class SocketConnection;
+
+
+class MessageReceiver
+{
+public:
+	virtual void onMessageReceived(SocketConnection* connection, const BaseMessage& baseMessage, char* buffer) = 0;
+};
+
+
+class SocketConnection
+{
+public:
+	SocketConnection(MessageReceiver* _receiver);
+	virtual ~SocketConnection();
+	virtual void start();
+	virtual void stop();
+	virtual void send(BaseMessage* _message);
+
+protected:
+	virtual void worker() = 0;
+
+	void socketRead(void* _to, size_t _bytes);
+	std::shared_ptr<tcp::socket> socket;
+
+//	boost::asio::ip::tcp::endpoint endpt;
+	std::atomic<bool> active_;
+	MessageReceiver* messageReceiver;
+	void getNextMessage();
+	boost::asio::io_service io_service;
+	tcp::resolver::iterator iterator;
+	std::thread* receiverThread;
+	mutable std::mutex mutex_;
+};
+
+
+class ClientConnection : public SocketConnection
+{
+public:
+	ClientConnection(MessageReceiver* _receiver, const std::string& _ip, size_t _port);
+
+protected:
+	virtual void worker();
+
+private:
+	std::string ip;
+	size_t port;
+};
+
+
+
+class ServerConnection : public SocketConnection
+{
+public:
+	ServerConnection(MessageReceiver* _receiver, std::shared_ptr<tcp::socket> _socket);
+
+protected:
+	virtual void worker();
+};
+
+
+
+
+#endif
+
+
+
+

common/queue.h

@@ -1,3 +1,6 @@
+#ifndef QUEUE_H
+#define QUEUE_H
+
 #include <queue>
 #include <thread>
 #include <mutex>
@@ -86,4 +89,6 @@ public:
 };
 
 
+#endif
+
 

svn-commit.tmp~

@@ -0,0 +1,4 @@
+
+-- Diese und die folgenden Zeilen werden ignoriert --
+
+M    client/serverConnection.cpp