xxx

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

Makefile

@@ -1,9 +1,9 @@
 VERSION = 0.01
 CC      = /usr/bin/g++
 CFLAGS  = -std=gnu++0x -Wall -Wno-unused-function -g -O3 -D_REENTRANT -DVERSION=\"$(VERSION)\"
-LDFLAGS = -lrt -lzmq -lpthread -lportaudio
+LDFLAGS = -lrt -lzmq -lpthread -lportaudio -lboost_system
 
-OBJ_SERVER = server.o
+OBJ_SERVER = blocking_tcp_echo_server.o
 BIN_SERVER = server
 OBJ_CLIENT = client.o stream.o chunk.o
 BIN_CLIENT = client

blocking_tcp_echo_client.cpp

@@ -83,7 +83,7 @@ int main(int argc, char* argv[])
 		std::cerr << "Exception: " << e.what() << "\n";
 	}
 
-  return 0;
+	return 0;
 }
 
 

blocking_tcp_echo_server.cpp

@@ -18,6 +18,12 @@
 #include <ctime>   // localtime
 #include <sstream> // stringstream
 #include <iomanip>
+#include <thread>
+#include <memory>
+#include "chunk.h"
+#include "timeUtils.h"
+#include "queue.h"
+
 
 using boost::asio::ip::tcp;
 
@@ -41,72 +47,157 @@ std::string return_current_time_and_date()
 }
 
 
-void session(socket_ptr sock)
+class Session
 {
-  try
+public:
-  {
+	Session(socket_ptr sock) : socket_(sock)
-    for (;;)
+	{
-    {
+	}
-/*      char data[max_length];
 
-      boost::system::error_code error;
+	void sender()
-      size_t length = sock->read_some(boost::asio::buffer(data), error);
+	{
-      if (error == boost::asio::error::eof)
+		try
-        break; // Connection closed cleanly by peer.
+		{
-      else if (error)
+			for (;;)
-        throw boost::system::system_error(error); // Some other error.
+			{
-*/
+				shared_ptr<WireChunk> chunk(chunks.pop());
-	  string response(return_current_time_and_date() + "\tHallo\n");
+				boost::system::error_code error;
-		boost::system::error_code error;
+				boost::asio::write(*socket_, boost::asio::buffer(chunk.get(), sizeof(WireChunk)), boost::asio::transfer_all(), error);
-		cout << response << "\n";
+				if (error == boost::asio::error::eof)
-		boost::asio::write(*sock, boost::asio::buffer(response), boost::asio::transfer_all(), error);
+					break; // Connection closed cleanly by peer.
-		if (error == boost::asio::error::eof)
+				else if (error)
-			break; // Connection closed cleanly by peer.
+					throw boost::system::system_error(error); // Some other error.
-		else if (error)
+			}
-			throw boost::system::system_error(error); // Some other error.
+		}
-		sleep(1);
+		catch (std::exception& e)
-//std::cout << return_current_time_and_date() << "\n";
+		{
-    }
+			std::cerr << "Exception in thread: " << e.what() << "\n";
-  }
+		}
-  catch (std::exception& e)
+	}
-  {
-    std::cerr << "Exception in thread: " << e.what() << "\n";
-  }
-}
 
-void server(boost::asio::io_service& io_service, unsigned short port)
+	void start()
+	{
+		senderThread = new thread(&Session::sender, this);
+//		readerThread.join();
+	}
+
+	void send(shared_ptr<WireChunk> chunk)
+	{
+		chunks.push(chunk);
+	}
+
+private:
+	socket_ptr socket_;
+	thread* senderThread;
+	Queue<shared_ptr<WireChunk>> chunks;
+};
+
+
+class Server
 {
-  tcp::acceptor a(io_service, tcp::endpoint(tcp::v4(), port));
+public:
-  for (;;)
+	Server(unsigned short port) : session(NULL), port_(port)
-  {
+	{
-    socket_ptr sock(new tcp::socket(io_service));
+	}
-    a.accept(*sock);
+
-	cout << "New connection: " << sock->remote_endpoint().address().to_string() << "\n";
+	void acceptor()
-    boost::thread t(boost::bind(session, sock));
+	{
-  }
+		tcp::acceptor a(io_service_, tcp::endpoint(tcp::v4(), port_));
-}
+		for (;;)
+		{
+			socket_ptr sock(new tcp::socket(io_service_));
+			a.accept(*sock);
+			cout << "New connection: " << sock->remote_endpoint().address().to_string() << "\n";
+			session = new Session(sock);
+			session->start();
+		}
+	}
+	
+	void send(shared_ptr<WireChunk> chunk)
+	{
+		if (session != 0)
+			session->send(chunk);
+	}
+
+	void start()
+	{
+		acceptThread = new thread(&Server::acceptor, this);
+	}
+
+	void stop()
+	{
+		acceptThread->join();
+	}
+
+private:
+	Session* session;
+	boost::asio::io_service io_service_;
+	unsigned short port_;
+	thread* acceptThread;
+};
+
 
 int main(int argc, char* argv[])
 {
-  try
+	try
-  {
+	{
-    if (argc != 2)
+		if (argc != 2)
-    {
+		{
-      std::cerr << "Usage: blocking_tcp_echo_server <port>\n";
+			std::cerr << "Usage: blocking_tcp_echo_server <port>\n";
-      return 1;
+			return 1;
-    }
+		}
 
-    boost::asio::io_service io_service;
 
-    using namespace std; // For atoi.
+		using namespace std; // For atoi.
-    server(io_service, atoi(argv[1]));
+		Server* server = new Server(atoi(argv[1]));
-  }
+		server->start();
-  catch (std::exception& e)
-  {
-    std::cerr << "Exception: " << e.what() << "\n";
-  }
 
-  return 0;
+		char c[2];
+		timeval tvChunk;
+		gettimeofday(&tvChunk, NULL);
+		long nextTick = getTickCount();
+
+		while (cin.good())
+		{
+			shared_ptr<WireChunk> chunk(new WireChunk());
+			for (size_t n=0; (n<WIRE_CHUNK_SIZE) && cin.good(); ++n)
+			{
+				c[0] = cin.get();
+				c[1] = cin.get();
+			    chunk->payload[n] = (int)c[0] + ((int)c[1] << 8);
+			}
+
+	//		if (!cin.good())
+	//			cin.clear();
+
+		    chunk->tv_sec = tvChunk.tv_sec;
+		    chunk->tv_usec = tvChunk.tv_usec;
+			server->send(chunk);
+
+		    addMs(tvChunk, WIRE_CHUNK_MS);
+			nextTick += WIRE_CHUNK_MS;
+			long currentTick = getTickCount();
+			if (nextTick > currentTick)
+			{
+				usleep((nextTick - currentTick) * 1000);
+			}
+			else
+			{
+				cin.sync();
+				gettimeofday(&tvChunk, NULL);
+				nextTick = getTickCount();
+			}
+		}
+		return 0;
+		server->stop();	
+	}
+	catch (std::exception& e)
+	{
+		std::cerr << "Exception: " << e.what() << "\n";
+	}
+
+	return 0;
 }
 
 
+

client.cpp

@@ -15,12 +15,14 @@
 #include <algorithm>
 #include <thread> 
 #include <portaudio.h>
+#include <boost/asio.hpp>
 
 #include "chunk.h"
 #include "utils.h"
 #include "stream.h"
 #include "zhelpers.hpp"
 
+using boost::asio::ip::tcp;
 using namespace std;
 
 
@@ -32,18 +34,43 @@ Stream* stream;
 
 void player() 
 {
-    zmq::context_t context (1);
+	try
-    zmq::socket_t subscriber (context, ZMQ_SUB);
+	{
-    subscriber.connect("tcp://192.168.0.2:123458");
+		boost::asio::io_service io_service;
+		tcp::resolver resolver(io_service);
+		tcp::resolver::query query(tcp::v4(), "192.168.0.2", "98765");
+		tcp::resolver::iterator iterator = resolver.resolve(query);
 
-    const char* filter = "";
+		while (true)
-    subscriber.setsockopt(ZMQ_SUBSCRIBE, filter, strlen(filter));
+		{
-    zmq::message_t update;
+			try
-    while (1)
+			{
-    {
+				tcp::socket s(io_service);
-        subscriber.recv(&update);
+				s.connect(*iterator);
-		stream->addChunk(new Chunk((WireChunk*)(update.data())));
+				boost::array<char, 128> buf;
-    }
+				boost::system::error_code error;
+
+				while (true)
+				{
+					WireChunk* wireChunk = new WireChunk();
+					size_t len = s.read_some(boost::asio::buffer(wireChunk, sizeof(WireChunk)));
+					if (error == boost::asio::error::eof)
+						break;
+
+					stream->addChunk(new Chunk(wireChunk));
+				}
+			}
+			catch (std::exception& e)
+			{
+				std::cerr << "Exception: " << e.what() << "\n";
+				usleep(100*1000);
+			}
+		}
+	}
+	catch (std::exception& e)
+	{
+		std::cerr << "Exception: " << e.what() << "\n";
+	}
 }