org.argeo.slc.webapp/src/main/webapp/argeo-ria-src/class/org/argeo/ria/remote/AmqClient.js

   1 qx.Class.define("org.argeo.ria.remote.AmqClient", {
   2
   3         extend : qx.core.Object,
   4         construct : function(){
   5                 this.base(arguments);
   6         },
   7         members : {
   8                 // The URI of the MessageListenerServlet
   9                 uri : '/org.argeo.slc.webapp/amq',
  10
  11                 // Polling. Set to true (default) if waiting poll for messages is needed
  12                 poll : true,
  13
  14                 // Poll delay. if set to positive integer, this is the time to wait in ms before
  15                 // sending the next poll after the last completes.
  16                 _pollDelay : 0,
  17
  18                 _first : true,
  19                 _pollEvent : function(first) {
  20                 },
  21                 _handlers : new Array(),
  22
  23                 _messages : 0,
  24                 _messageQueue : '',
  25                 _queueMessages : 0,
  26
  27                 _messageHandler : function(response) {
  28                         var doc = response.getContent();
  29                         var NSMap = {slc:"http://argeo.org/projects/slc/schemas"};
  30                         var messages = org.argeo.ria.util.Element.selectNodes(doc, "//response", NSMap);
  31                         //this.info("Received " + messages.length + " messages");
  32                         for(var i=0;i<messages.length;i++){
  33                                 var id = messages[i].getAttribute("id");
  34                                 var slcExec = org.argeo.ria.util.Element.selectSingleNode(messages[i], "slc:slc-execution", NSMap);
  35                                 if(id && this._handlers[id] && slcExec){
  36                                         this._handlers[id](slcExec);
  37                                 }
  38                         }
  39                 },
  40
  41                 startBatch : function() {
  42                         this._queueMessages++;
  43                 },
  44
  45                 endBatch : function() {
  46                         this._queueMessages--;
  47                         if (this._queueMessages == 0 && this._messages > 0) {
  48                                 var body = this._messageQueue;
  49                                 this._messageQueue = '';
  50                                 this._messages = 0;
  51                                 this._queueMessages++;
  52                                 var request = new qx.io.remote.Request(this.uri, "POST", "text/plain");
  53                                 request.addListener("completed", this.endBatch, this);
  54                                 request.send();
  55                         }
  56                 },
  57
  58                 _pollHandler : function(response) {
  59                         this.startBatch();
  60                         try {
  61                                 this._messageHandler(response);
  62                                 this._pollEvent(this._first);
  63                                 this._first = false;
  64                         } catch (e) {
  65                                 alert(e);
  66                         }
  67                         this.endBatch();
  68
  69                         if (this._pollDelay > 0)
  70                                 qx.event.Timer.once(this._sendPoll, this, this._pollDelay);
  71                         else
  72                                 this._sendPoll();
  73                 },
  74
  75                 _sendPoll : function(request) {
  76                         var request = new qx.io.remote.Request(this.uri, "GET", "application/xml");
  77                         request.addListener("completed", this._pollHandler, this);
  78                         request.send();
  79                 },
  80
  81                 // Add a function that gets called on every poll response, after all received
  82                 // messages have been handled.  The poll handler is past a boolean that indicates
  83                 // if this is the first poll for the page.
  84                 addPollHandler : function(func) {
  85                         var old = this._pollEvent;
  86                         this._pollEvent = function(first) {
  87                                 old(first);
  88                                 func(first);
  89                         }
  90                 },
  91
  92                 // Send a JMS message to a destination (eg topic://MY.TOPIC).  Message should be xml or encoded
  93                 // xml content.
  94                 sendMessage : function(destination, message) {
  95                         this._sendMessage(destination, message, 'send');
  96                 },
  97
  98                 // Listen on a channel or topic.   handler must be a function taking a message arguement
  99                 addListener : function(id, destination, handler, context) {
 100                         this._handlers[id] = qx.lang.Function.bind(handler, context);
 101                         this._sendMessage(destination, id, 'listen');
 102                 },
 103
 104                 // remove Listener from channel or topic.
 105                 removeListener : function(id, destination) {
 106                         this._handlers[id] = null;
 107                         this._sendMessage(destination, id, 'unlisten');
 108                 },
 109
 110                 _sendMessage : function(destination, message, type) {
 111                         if (this._queueMessages > 0) {
 112                                 if (this._messages == 0) {
 113                                         this._messageQueue = 'destination=' + destination
 114                                                         + '&message=' + message + '&type=' + type;
 115                                 } else {
 116                                         this._messageQueue += '&d' + this._messages + '='
 117                                                         + destination + '&m' + this._messages + '='
 118                                                         + message + '&t' + this._messages + '=' + type;
 119                                 }
 120                                 this._messages++;
 121                         } else {
 122                                 this.startBatch();
 123                                 var req = new qx.io.remote.Request(this.uri, "POST", "text/plain");
 124                                 req.setParameter("destination", destination);
 125                                 req.setParameter("message", message);
 126                                 req.setParameter("type", type);
 127                                 req.addListener("completed", this.endBatch, this);
 128                                 req.send();
 129                         }
 130                 },
 131
 132                 startPolling : function() {
 133                         if (this.poll){
 134                                 var req = new qx.io.remote.Request(this.uri, "GET", "application/xml");
 135                                 req.setParameter("timeout", "10");
 136                                 req.addListener("completed", this._pollHandler, this);
 137                                 req.send();
 138                         }
 139                 }
 140         }
 141 });