ProducerConsumer.cpp/NetworkService.h at main · tty168/ProducerConsumer.cpp · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
#ifndef _NetworkService_H
#define _NetworkService_H

//
// Copyright (c) 2002 by Ted T. Yuan.
//
// Permission is granted to use this code without restriction as long as this copyright notice appears in all source files.
//
#include <stdio.h>
#include <string.h>
#include <iostream>
#ifdef _DEBUG
#pragma warning (disable: 4786 4788)
#endif
#include <string>
#include <vector>
#if defined(WIN32)
#include <windows.h>
#else
#include <unistd.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <resolv.h>
#include <netdb.h>
#endif
#include <ThreadPool.h>
#include <socket_utils.h>

#ifndef SPACE_YIN
#define SPACE_YIN yin
#endif

namespace SPACE_YIN {

struct NetworkService;

struct SocketHandler
{
	NetworkService& ns_;
	int client_;
	SocketHandler(NetworkService& ns, int socket)
		: ns_(ns), client_(socket) { }

	~SocketHandler() { }

	void operator()();
};

struct NetworkService
{
	std::string serviceId;
	size_t lengthWaitList;
	size_t numThreads;
	unsigned short sourcePort;

	NetworkService(std::string serviceId_in,
		int lengthWaitList_in, int numThreads_in, unsigned short sourcePort_in)
		: serviceId(serviceId_in), lengthWaitList(lengthWaitList_in),
		numThreads(numThreads_in), sourcePort(sourcePort_in) {}

	// sub-classes over-write this method and close socket
	virtual void ServiceHandler(int socket) = 0;

	virtual bool stop() { return false; }

	void operator()()
	{
		int server = createServerSocket(sourcePort, lengthWaitList+1);
		if(server < 0)
		{
			std::cout << "Can't bind to port " << sourcePort << std::endl;
			return;
		}

		RunPool<SocketHandler> pool(numThreads);
		ThreadPool<SocketHandler> thpool(pool,
			numThreads > 256 ? 256 : numThreads);
		// start the pool...
		boost::thread thrd(thpool);

		std::cout << "Ready... service at port " << sourcePort << std::endl;
		while (!stop())
		{
			struct   sockaddr_in pin;
			int source = acceptClientSocket(server, pin);
			if(source < 0) continue;

			// put it on the heap...
			SocketHandler * linkage = new SocketHandler(*this, source);

			thpool.execute(linkage);
		}

		thrd.join();
		CloseConnection(server);
		std::cout << "Shutting down network service " <<
			serviceId << " at port " << sourcePort << std::endl;
	}
};

void SocketHandler::operator()()
{
	try
	{
		ns_.ServiceHandler(client_);
	} catch (...) {}
	delete this;
}

}

#endif