Mercurial > projects > ldc

comparison tango/tango/net/cluster/tina/ClusterServer.d @ 132:1700239cab2e trunk

Find changesets by keywords (author, files, the commit message), revision number or hash, or revset expression.
[svn r136] MAJOR UNSTABLE UPDATE!!! Initial commit after moving to Tango instead of Phobos. Lots of bugfixes... This build is not suitable for most things.
author lindquist
date Fri, 11 Jan 2008 17:57:40 +0100
parents
children
comparison
equal deleted inserted replaced
131:5825d48b27d1 132:1700239cab2e
1 /*******************************************************************************
2
3 copyright: Copyright (c) 2004 Kris Bell. All rights reserved
4
5 license: BSD style: $(LICENSE)
6
7 version: July 2004: Initial release
8
9 author: Kris
10
11 *******************************************************************************/
12
13 module tango.net.cluster.tina.ClusterServer;
14
15 private import tango.core.Thread;
16
17 private import tango.util.ArgParser;
18
19 private import tango.net.Socket,
20 tango.net.InternetAddress;
21
22 private import tango.net.cluster.tina.Cluster,
23 tango.net.cluster.tina.RollCall;
24
25 private import tango.net.cluster.NetworkRegistry;
26
27 private import tango.net.cluster.tina.util.ServerThread;
28
29 package import tango.net.cluster.tina.util.AbstractServer;
30
31 /******************************************************************************
32
33 Extends the AbstractServer to glue cluster support together.
34
35 ******************************************************************************/
36
37 abstract class ClusterServer : AbstractServer
38 {
39 package char[] name;
40 package Cluster cluster;
41 package IChannel channel;
42 package RollCall rollcall;
43
44 /**********************************************************************
45
46 Concrete server must expose a service handler
47
48 **********************************************************************/
49
50 abstract void service (IConduit conduit);
51
52 /**********************************************************************
53
54 Construct this server with the requisite attributes. The
55 'bind' address is the local address we'll be listening on,
56 'threads' represents the number of socket-accept threads,
57 and backlog is the number of "simultaneous" connection
58 requests that a socket layer will buffer on our behalf.
59
60 We also set up a listener for client discovery-requests,
61 and lastly, we tell active clients that we're available
62 for work. Clients should be listening on the appropriate
63 channel for an instance of the RollCall payload.
64
65 **********************************************************************/
66
67 this (char[] name, InternetAddress bind, Logger logger)
68 {
69 this.name = name;
70
71 super (bind, 1, 50, logger);
72
73 // hook into the cluster as a server
74 cluster = new Cluster (logger);
75 }
76
77 /**********************************************************************
78
79 **********************************************************************/
80
81 void enroll (IMessage task)
82 {
83 NetworkRegistry.shared.enroll (task);
84 }
85
86 /**********************************************************************
87
88 Start the server
89
90 Note that we hijack the calling thread, and use it to
91 generate a hearbeat. The hearbeat has two functions: it
92 tells all clients when this server starts, and it tells
93 them we're still alive. The latter is important if, for
94 example, a client request to this server had timed-out
95 due to the server being too busy. In such a case, the
96 client will mark the server as being unavailable, and
97 the heartbeat will presumably revert that.
98
99 It would also be useful to monitor the GC from here.
100
101 **********************************************************************/
102
103 void start (RollCall id, bool reuse=false)
104 {
105 super.start (reuse);
106
107 // configure an identity for ourselves
108 id.addr = Socket.hostName ~ ':' ~ localAddress.toPortString;
109 this.rollcall = id;
110
111 // clients are listening on this channel ...
112 channel = cluster.createChannel ("cluster.server.advertise");
113
114 // ... and listen for subsequent server.advertise requests
115 channel.createBulletinConsumer (&notify);
116
117 while (true)
118 {
119 getLogger.trace ("heartbeat");
120 channel.broadcast (rollcall);
121 Thread.sleep (30.0);
122 }
123 }
124
125 /**********************************************************************
126
127 Return a text string identifying this server
128
129 **********************************************************************/
130
131 char[] getProtocol ()
132 {
133 return name;
134 }
135
136 /**********************************************************************
137
138 Return a text string identifying this server
139
140 **********************************************************************/
141
142 override char[] toString ()
143 {
144 return "cluster::" ~ name;
145 }
146
147 /**********************************************************************
148
149 Create a ServerSocket instance.
150
151 **********************************************************************/
152
153 override ServerSocket createSocket (InternetAddress bind, int backlog, bool reuse=false)
154 {
155 return new ServerSocket (bind, backlog, reuse);
156 }
157
158 /**********************************************************************
159
160 Create a ServerThread instance. This can be overridden to
161 create other thread-types, perhaps with additional thread-
162 level data attached.
163
164 **********************************************************************/
165
166 override void createThread (ServerSocket socket)
167 {
168 new ServerThread (this, socket);
169 }
170
171 /**********************************************************************
172
173 Interface method that's invoked when a client is making
174 discovery requests. We just send back our identity in a
175 reply
176
177 **********************************************************************/
178
179 private void notify (IEvent event)
180 {
181 scope input = new RollCall;
182 event.get (input);
183
184 // if this is a request, reply with our identity
185 if (input.type is input.Request)
186 channel.broadcast (rollcall);
187 }
188 }
189