Controller


1   // Copyright (c) 2000 Just Objects B.V. <just@justobjects.nl>
2   // Distributable under LGPL license. See terms of license at gnu.org.
3   
4   package nl.justobjects.pushlet.core;
5   
6   import java.io.IOException  ;
7   
8   /**
9    * Handles servlet requests from client.
10   *
11   * @version $Id: Controller.java,v 1.8 2005/02/28 15:58:05 justb Exp $
12   * @author Just van den Broecke - Just Objects &copy;
13   **/
14  public class Controller implements Protocol, ConfigDefs {
15  
16      private Session session;
17  
18      public Controller(Session theSession) {
19          session = theSession;
20      }
21  
22      /** Handle command. */
23      public void doCommand(Command aCommand) {
24          try {
25              // Update lease time to live
26              session.kick();
27  
28              // Set remote IP address of client
29              session.setAddress(aCommand.httpReq.getRemoteAddr());
30  
31              debug("doCommand() event=" + aCommand.reqEvent);
32  
33              // Get event type
34              String   eventType = aCommand.reqEvent.getEventType();
35  
36              // Determine action based on event type
37              if (eventType.equals(Protocol.E_REFRESH)) {
38                  // Pull/poll mode clients that refresh
39                  doRefresh(aCommand);
40              } else if (eventType.equals(Protocol.E_SUBSCRIBE)) {
41                  // Subscribe
42                  doSubscribe(aCommand);
43              } else if (eventType.equals(Protocol.E_UNSUBSCRIBE)) {
44                  // Unsubscribe
45                  doUnsubscribe(aCommand);
46              } else if (eventType.equals(Protocol.E_JOIN)) {
47                  // Join
48                  doJoin(aCommand);
49              } else if (eventType.equals(Protocol.E_JOIN_LISTEN)) {
50                  // Join and listen (for simple and e.g. REST apps)
51                  doJoinListen(aCommand);
52              } else if (eventType.equals(Protocol.E_LEAVE)) {
53                  // Leave
54                  doLeave(aCommand);
55              } else if (eventType.equals(Protocol.E_HEARTBEAT)) {
56                  // Heartbeat mainly to do away with browser "busy" cursor
57                  doHeartbeat(aCommand);
58              } else if (eventType.equals(Protocol.E_PUBLISH)) {
59                  // Publish event
60                  doPublish(aCommand);
61              } else if (eventType.equals(Protocol.E_LISTEN)) {
62                  // Listen to pushed events
63                  doListen(aCommand);
64              }
65  
66              // Handle response back to client
67              if (eventType.endsWith(Protocol.E_LISTEN) ||
68                      eventType.equals(Protocol.E_REFRESH)) {
69                  // Data channel events
70                  // Loops until refresh or connection closed
71                  getSubscriber().fetchEvents(aCommand);
72  
73              } else {
74                  // Send response for control commands
75                  sendControlResponse(aCommand);
76              }
77  
78          } catch (Throwable   t) {
79              warn("Exception in doCommand(): " + t);
80              t.printStackTrace();
81          }
82      }
83  
84      public String   toString() {
85          return session.toString();
86      }
87  
88      /** Handle heartbeat event. */
89      protected void doHeartbeat(Command aCommand) {
90  
91          // Set heartbeat acknowledgement to client
92          aCommand.setResponseEvent(new Event(E_HEARTBEAT_ACK));
93      }
94  
95      /** Handle Join request. */
96      protected void doJoin(Command aCommand) throws IOException   {
97  
98          Event responseEvent = null;
99  
100         try {
101 
102             session.start();
103 
104             // Determine format for encoding Events to client.
105             // Default assume a userAgent window on the other end.
106             String   format = aCommand.reqEvent.getField(P_FORMAT, FORMAT_JAVASCRIPT);
107 
108             session.setFormat(format);
109             responseEvent = new Event(E_JOIN_ACK);
110 
111             // Set unique subscriber id and encoding format
112             responseEvent.setField(P_ID, session.getId());
113             responseEvent.setField(P_FORMAT, format);
114             info("joined");
115         } catch (Throwable   t) {
116             session.stop();
117             responseEvent = new Event(E_NACK);
118             responseEvent.setField(P_ID, session.getId());
119             responseEvent.setField(P_REASON, "unexpected error: " + t);
120             warn("doJoin() error: " + t);
121             t.printStackTrace();
122         } finally {
123             // Always set response event in command
124             aCommand.setResponseEvent(responseEvent);
125         }
126 
127     }
128 
129     /** Handle JoinListen request. */
130     protected void doJoinListen(Command aCommand) throws IOException   {
131 
132         // Basically bundles a join and a listen
133         // This request is handly for simple apps that
134         // need to do a single request to get events immediately
135         // For example in RESTful apps.
136 
137         // First do regular join
138         doJoin(aCommand);
139         if (!aCommand.getResponseEvent().getEventType().equals(E_NACK)) {
140             // If successful do the listen
141             doListen(aCommand);
142             if (!aCommand.getResponseEvent().getEventType().equals(E_NACK)) {
143                 // If still ok do the listen ack
144                 aCommand.getResponseEvent().setField(P_EVENT, E_JOIN_LISTEN_ACK);
145             }
146         }
147     }
148 
149     /** Handle Leave request. */
150     protected void doLeave(Command aCommand) throws IOException   {
151 
152         Event responseEvent = null;
153 
154         try {
155             // Remove all subscriptions
156             getSubscriber().bailout();
157 
158             // Prepare acknowledgement
159             responseEvent = new Event(E_LEAVE_ACK);
160 
161             // Set unique subscriber id
162             responseEvent.setField(P_ID, session.getId());
163             info("left");
164         } catch (Throwable   t) {
165             responseEvent = new Event(E_NACK);
166             responseEvent.setField(P_ID, session.getId());
167             responseEvent.setField(P_REASON, "unexpected error: " + t);
168             warn("doLeave() error: " + t);
169             t.printStackTrace();
170         } finally {
171             // Always set response event in command
172             aCommand.setResponseEvent(responseEvent);
173         }
174 
175     }
176 
177     /** Handle Listen request. */
178     protected void doListen(Command aCommand) throws IOException   {
179 
180 
181         String   mode = MODE_STREAM;
182         // Should we always force "pull" mode ?
183         if (Config.getBoolProperty(LISTEN_FORCE_PULL_ALL)) {
184             mode = MODE_PULL;
185         } else {
186             // Determine optimal mode determined by parameter and/or user agent
187             // Mode param determines how events are transfered to the client
188 
189             // In "stream" mode, a stream of events is sent, i.e. the document
190             // is neverending. In "pull" or "poll" mode a complete document is returned
191             // ending with a request to refresh.
192             mode = aCommand.reqEvent.getField(P_MODE, MODE_STREAM);
193 
194             String   userAgent = aCommand.httpReq.getHeader("User-Agent");
195             if (userAgent != null) {
196                 userAgent = userAgent.toLowerCase();
197                 for (int i = 0; i < session.FORCED_PULL_AGENTS.length; i++) {
198                     if ((userAgent.indexOf(session.FORCED_PULL_AGENTS[i]) != -1)) {
199                         info("Forcing pull mode for agent=" + userAgent);
200                         mode = MODE_PULL;
201                         break;
202                     }
203                 }
204             } else {
205                 userAgent = "unknown";
206             }
207         }
208 
209         getSubscriber().setMode(mode);
210 
211         // Prepare acknowledgement
212         Event listenAckEvent = new Event(E_LISTEN_ACK);
213 
214         // Add subscription(s) if subject(s) specified
215         String   subject = aCommand.reqEvent.getField(P_SUBJECT);
216         if (subject != null) {
217             // Optional label for subscription
218             String   label = aCommand.reqEvent.getField(Protocol.P_SUBSCRIPTION_LABEL);
219 
220             // Add a subscription
221             Subscription subscription = getSubscriber().addSubscription(subject, label);
222 
223             // Add subscription id and optional label to listen-ack event
224             listenAckEvent.setField(P_SUBSCRIPTION_ID, subscription.getId());
225             if (label != null) {
226                 listenAckEvent.setField(P_SUBSCRIPTION_LABEL, label);
227             }
228         }
229 
230         // Set unique subscriber id, push mode and encoding format
231         listenAckEvent.setField(P_ID, session.getId());
232         listenAckEvent.setField(P_MODE, mode);
233         listenAckEvent.setField(P_FORMAT, session.getFormat());
234 
235         // Activate the subscriber
236         getSubscriber().activate();
237 
238         // Enqueue listen ack event on data channel
239         aCommand.setResponseEvent(listenAckEvent);
240 
241         info("Listening mode=" + mode + " userAgent=" + session.getUserAgent());
242 
243     }
244 
245     /** Handle Publish request. */
246     protected void doPublish(Command aCommand) {
247         Event responseEvent = null;
248 
249         try {
250             String   subject = aCommand.reqEvent.getField(Protocol.P_SUBJECT);
251             if (subject == null) {
252                 // Return error response
253                 responseEvent = new Event(E_NACK);
254                 responseEvent.setField(P_ID, session.getId());
255                 responseEvent.setField(P_REASON, "no subject provided");
256             } else {
257                 aCommand.reqEvent.setField(P_FROM, session.getId());
258                 aCommand.reqEvent.setField(P_EVENT, E_DATA);
259 
260                 // Event may be targeted to specific user (p_to field)
261                 String   to = aCommand.reqEvent.getField(P_TO);
262                 if (to != null) {
263                     Dispatcher.getInstance().unicast(aCommand.reqEvent, to);
264                 } else {
265                     // No to: multicast
266                     debug("doPublish() event=" + aCommand.reqEvent);
267                     Dispatcher.getInstance().multicast(aCommand.reqEvent);
268                 }
269 
270                 // Acknowledge
271                 responseEvent = new Event(E_PUBLISH_ACK);
272             }
273 
274         } catch (Throwable   t) {
275             responseEvent = new Event(E_NACK);
276             responseEvent.setField(P_ID, session.getId());
277             responseEvent.setField(P_REASON, "unexpected error: " + t);
278             warn("doPublish() error: " + t);
279             t.printStackTrace();
280         } finally {
281             // Always set response event in command
282             aCommand.setResponseEvent(responseEvent);
283         }
284     }
285 
286     /** Handle refresh event. */
287     protected void doRefresh(Command aCommand) {
288         // Set ack
289         aCommand.setResponseEvent(new Event(E_REFRESH_ACK));
290     }
291 
292     /** Handle Subscribe request. */
293     protected void doSubscribe(Command aCommand) throws IOException   {
294 
295         Event responseEvent = null;
296         try {
297             String   subject = aCommand.reqEvent.getField(Protocol.P_SUBJECT);
298             Subscription subscription = null;
299             if (subject == null) {
300                 // Return error response
301                 responseEvent = new Event(E_NACK);
302                 responseEvent.setField(P_ID, session.getId());
303                 responseEvent.setField(P_REASON, "no subject provided");
304             } else {
305 
306                 String   label = aCommand.reqEvent.getField(Protocol.P_SUBSCRIPTION_LABEL);
307                 subscription = getSubscriber().addSubscription(subject, label);
308 
309                 // Acknowledge
310                 responseEvent = new Event(E_SUBSCRIBE_ACK);
311                 responseEvent.setField(P_ID, session.getId());
312                 responseEvent.setField(P_SUBJECT, subject);
313                 responseEvent.setField(P_SUBSCRIPTION_ID, subscription.getId());
314                 if (label != null) {
315                     responseEvent.setField(P_SUBSCRIPTION_LABEL, label);
316                 }
317                 info("subscribed to " + subject + " sid=" + subscription.getId());
318             }
319 
320         } catch (Throwable   t) {
321             responseEvent = new Event(E_NACK);
322             responseEvent.setField(P_ID, session.getId());
323             responseEvent.setField(P_REASON, "unexpected error: " + t);
324             warn("doSubscribe() error: " + t);
325             t.printStackTrace();
326         } finally {
327             // Always set response event in command
328             aCommand.setResponseEvent(responseEvent);
329         }
330     }
331 
332     /** Handle Unsubscribe request. */
333     protected void doUnsubscribe(Command aCommand) throws IOException   {
334 
335 
336         Event responseEvent = null;
337         try {
338             String   subscriptionId = aCommand.reqEvent.getField(Protocol.P_SUBSCRIPTION_ID);
339             if (subscriptionId == null) {
340                 // Unsuscbribe all
341                 getSubscriber().removeSubscriptions();
342                 responseEvent = new Event(E_UNSUBSCRIBE_ACK);
343                 responseEvent.setField(P_ID, session.getId());
344                 info("unsubscribed all");
345             } else {
346                 // Subscription id provided: remove Subscription
347                 Subscription subscription = getSubscriber().removeSubscription(subscriptionId);
348                 if (subscription == null) {
349                     // Unknown subscription id: return error response
350                     responseEvent = new Event(E_NACK);
351                     responseEvent.setField(P_ID, session.getId());
352                     responseEvent.setField(P_REASON, "no subscription for sid=" + subscriptionId);
353                     warn("unsubscribe: no subscription for sid=" + subscriptionId);
354                 } else {
355                     // OK return ack
356                     responseEvent = new Event(E_UNSUBSCRIBE_ACK);
357                     responseEvent.setField(P_ID, session.getId());
358                     responseEvent.setField(P_SUBSCRIPTION_ID, subscription.getId());
359                     responseEvent.setField(P_SUBJECT, subscription.getSubject());
360                     if (subscription.getLabel() != null) {
361                         responseEvent.setField(P_SUBSCRIPTION_LABEL, subscription.getLabel());
362                     }
363                     info("unsubscribed sid= " + subscriptionId);
364                 }
365             }
366         } catch (Throwable   t) {
367             responseEvent = new Event(E_NACK);
368             responseEvent.setField(P_ID, session.getId());
369             responseEvent.setField(P_REASON, "unexpected error: " + t);
370             warn("doUnsubscribe() error: " + t);
371             t.printStackTrace();
372         } finally {
373             // Always set response event in command
374             aCommand.setResponseEvent(responseEvent);
375         }
376     }
377 
378     public Subscriber getSubscriber() {
379         return session.getSubscriber();
380     }
381 
382     /** Send response on the control channel. */
383     protected void sendControlResponse(Command aCommand) {
384         try {
385 
386             // Try to prevent caching in any form.
387             aCommand.sendResponseHeaders();
388 
389             // Let clientAdapter determine how to send event
390             aCommand.getClientAdapter().start();
391 
392             // Push to client through client adapter
393             aCommand.getClientAdapter().push(aCommand.getResponseEvent());
394 
395             // One shot response
396             aCommand.getClientAdapter().stop();
397         } catch (Throwable   t) {
398             getSubscriber().bailout();
399             return;
400         }
401     }
402 
403 
404     /** Info. */
405     protected void info(String   s) {
406         session.info("[Controller] " + s);
407     }
408 
409     /** Exceptional print util. */
410     protected void warn(String   s) {
411         session.warn("[Controller] " + s);
412     }
413 
414     /** Exceptional print util. */
415     protected void debug(String   s) {
416         session.debug("[Controller] " + s);
417     }
418 
419 
420 }
421 
422 /*
423  * $Log: Controller.java,v $
424  * Revision 1.8  2005/02/28 15:58:05  justb
425  * added SimpleListener example
426  *
427  * Revision 1.7  2005/02/28 13:05:59  justb
428  * introduced join-listen protocol service
429  *
430  * Revision 1.6  2005/02/28 12:45:59  justb
431  * introduced Command class
432  *
433  * Revision 1.5  2005/02/28 09:14:55  justb
434  * sessmgr/dispatcher factory/singleton support
435  *
436  * Revision 1.4  2005/02/25 15:13:00  justb
437  * session id generation more robust
438  *
439  * Revision 1.3  2005/02/21 16:59:06  justb
440  * SessionManager and session lease introduced
441  *
442  * Revision 1.2  2005/02/21 12:32:28  justb
443  * fixed publish event in Controller
444  *
445  * Revision 1.1  2005/02/21 11:50:46  justb
446  * ohase1 of refactoring Subscriber into Session/Controller/Subscriber
447  *
448 
449  *
450  */
451
A to Z: JavaDoc & Examples Daily Java News & Articles Open Source Projects Open Source Codes Free Computer Books Remove Frame
Popular Tags