modules/bayeux/java/org/apache/tomcat/bayeux/ClientImpl.java - tomcat80 - Git at Google

 /*
  * Licensed to the Apache Software Foundation (ASF) under one or more
  * contributor license agreements.  See the NOTICE file distributed with
  * this work for additional information regarding copyright ownership.
  * The ASF licenses this file to You under the Apache License, Version 2.0
  * (the "License"); you may not use this file except in compliance with
  * the License.  You may obtain a copy of the License at
  *
  *      http://www.apache.org/licenses/LICENSE-2.0
  *
  * Unless required by applicable law or agreed to in writing, software
  * distributed under the License is distributed on an "AS IS" BASIS,
  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
 package org.apache.tomcat.bayeux;

 import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
 import java.util.Queue;
 import java.util.concurrent.ConcurrentLinkedQueue;

 import org.apache.catalina.comet.CometEvent;
 import org.json.JSONObject;
 import org.apache.cometd.bayeux.Bayeux;
 import org.apache.cometd.bayeux.Client;
 import org.apache.cometd.bayeux.Listener;
 import org.apache.cometd.bayeux.Message;
 import org.apache.juli.logging.Log;
 import org.apache.juli.logging.LogFactory;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.HashMap;
 import java.util.ArrayList;

 public class ClientImpl implements Client {

     public static final int SUPPORT_CALLBACK_POLL = 0x1;
     public static final int SUPPORT_LONG_POLL = 0x2;

     public static final String COMET_EVENT_ATTR = "org.apache.cometd.bayeux.client";

     private static final Log log = LogFactory.getLog(ClientImpl.class);

     protected static LinkedList<Message> EMPTY_LIST = new LinkedList<Message>();
     /**
      * queued message for remote clients.
      */
     protected LinkedList<Message> messages = null;

     /**
      *
      */
     protected Queue<CometEvent> events = new LinkedList<CometEvent>();

     /**
      * Unique id representing this client
      */
     protected String id;

     /**
      * supported connection types, defaults to long-polling
      */
     protected int supportedConnTypes = SUPPORT_LONG_POLL | SUPPORT_CALLBACK_POLL;

     /**
      * The desired connection type
      */
     protected int desirectConnType = SUPPORT_LONG_POLL;

     /**
      * Does this client use json-comment-filtered messages
      */
     protected boolean useJsonFiltered = false;

     /**
      * Same JVM clients, get local=true
      */
     protected boolean local;

     /**
      * The callback object for local clients
      */
     protected Listener listener;

     protected AtomicInteger nrofsubscriptions = new AtomicInteger(0);

     protected ClientImpl(String id, boolean local) {
         this.id = id;
         this.local = local;
         if (!local) messages = new LinkedList<Message>();
     }

     protected ClientImpl(String id, CometEvent event) {
         this(id,false);
         events = new ConcurrentLinkedQueue<CometEvent>();
         addCometEvent(event);
     }

     public synchronized void deliver(Message message) {
         deliverInternal(null,new MessageImpl[] {(MessageImpl)message});
     }

     public synchronized void deliver(Message[] message) {
         deliverInternal(null,message);
     }

     protected synchronized void deliverInternal(ChannelImpl channel, MessageImpl message) {
         deliverInternal(channel,new MessageImpl[] {message});
     }

     protected synchronized void deliverInternal(ChannelImpl channel, Message[] msgs) {
         if (isLocal()) {
             //local clients must have a listener
             ArrayList<Message> list = new ArrayList<Message>();
             for (int i=0; msgs!=null && i<msgs.length; i++) {
                 //dont deliver to ourselves
                 if (this!=msgs[i].getClient()) list.add(msgs[i]);
             }
             if (getListener() != null && list.size()>0) {
                 getListener().deliver(list.toArray(new Message[0]));
             }
         } else {
             for (int i=0; msgs!=null && i<msgs.length; i++) {
                 MessageImpl message = (MessageImpl)msgs[i];
                 if (this==message.getClient()) {
                     //dont deliver to ourself
                     continue;
                 }
                 //we are not implementing forever responses, if the client is connected
                 //then we will fire off the message
                 //first we check to see if we have any existing connections we can piggy back on
                 CometEvent event = events.poll();
                 boolean delivered = false;
                 //TODO TODO - check on thread safety, for writing and for getting last request.
                 if (event!=null) {
                     synchronized (event) {
                         RequestBase rq = (RequestBase)event.getHttpServletRequest().getAttribute(RequestBase.LAST_REQ_ATTR);
                         if (rq!=null) {
                             Map map = new HashMap();
                             try {
                                 map.put(Bayeux.CHANNEL_FIELD,message.getChannel().getId());
                                 map.put(Bayeux.DATA_FIELD,message);
                                 JSONObject json = new JSONObject(map);
                                 if (log.isDebugEnabled()) {
                                     log.debug("Message instantly delivered to remote client["+this+"] message:"+json);
                                 }
                                 rq.addToDeliveryQueue(this, json);
                                 //deliver the batch
                                 if (i==(msgs.length-1)) {
                                     rq.deliver(event, this);
                                     event.close(); //todo, figure out a better way, this means only one message gets delivered
                                     removeCometEvent(event); //and delivered instantly
                                 }
                                 delivered = true;
                             } catch (Exception x) {
                                 log.error(x);
                             }
                         }
                     }
                 }
                 if (!delivered) {
                     if (log.isDebugEnabled()) {
                         log.debug("Message added to queue for remote client["+this+"] message:"+message);
                     }
                     //queue the message for the next round
                     messages.add(message);
                 }
             }
         }
     }

     public String getId() {
         return this.id;
     }

     protected Listener getListener() {
         return listener;
     }

     public boolean hasMessages() {
         if (isLocal()) return false;
         else {
             return messages.size() > 0;
         }
     }

     public boolean isLocal() {
         return local;
     }

     public int getSupportedConnTypes() {
         return supportedConnTypes;
     }

     public int getDesirectConnType() {
         return desirectConnType;
     }

     public boolean useJsonFiltered() {
         return useJsonFiltered;
     }

     public void setListener(Listener listener) {
         this.listener = listener;
     }

     public void setSupportedConnTypes(int supportedConnTypes) {
         this.supportedConnTypes = supportedConnTypes;
     }

     public void setUseJsonFiltered(boolean useJsonFiltered) {
         this.useJsonFiltered = useJsonFiltered;
     }

     public void setDesirectConnType(int desirectConnType) {
         this.desirectConnType = desirectConnType;
     }

     public boolean supportsCallbackPoll() {
         return (supportedConnTypes & SUPPORT_CALLBACK_POLL) == SUPPORT_CALLBACK_POLL;
     }

     public boolean supportsLongPoll() {
         return (supportedConnTypes & SUPPORT_LONG_POLL) == SUPPORT_LONG_POLL;
     }

     public synchronized List<Message> takeMessages() {
         if (isLocal()) return null;
         if (messages.size()==0) return EMPTY_LIST;
         List result = new LinkedList(messages);
         messages.clear();
         return result;
     }

     public String toString() {
         StringBuilder buf = new StringBuilder(super.toString());
         buf.append(" id=").append(getId());
         return buf.toString();
     }

     public boolean isSubscribed() {
         return nrofsubscriptions.get()>0;
     }

     protected synchronized boolean addCometEvent(CometEvent event) {
         boolean result = false;
         if (!events.contains(event)) {
             events.add(event);
             result = true;
         }
         event.getHttpServletRequest().setAttribute(COMET_EVENT_ATTR,this);
         return result;
     }

     protected synchronized boolean removeCometEvent(CometEvent event) {
         boolean result = events.remove(event);
         event.getHttpServletRequest().removeAttribute(COMET_EVENT_ATTR);
         return result;
     }


     protected void subscribed(ChannelImpl ch) {
         nrofsubscriptions.addAndGet(1);
     }

     protected void unsubscribed(ChannelImpl ch) {
         nrofsubscriptions.addAndGet(-1);
     }

     public void startBatch(){
         //noop until improved
     }
     public void endBatch() {
         //noop until improved
     }

 }
	/*
	* Licensed to the Apache Software Foundation (ASF) under one or more
	* contributor license agreements. See the NOTICE file distributed with
	* this work for additional information regarding copyright ownership.
	* The ASF licenses this file to You under the Apache License, Version 2.0
	* (the "License"); you may not use this file except in compliance with
	* the License. You may obtain a copy of the License at
	*
	* http://www.apache.org/licenses/LICENSE-2.0
	*
	* Unless required by applicable law or agreed to in writing, software
	* distributed under the License is distributed on an "AS IS" BASIS,
	* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
	* See the License for the specific language governing permissions and
	* limitations under the License.
	*/
	package org.apache.tomcat.bayeux;

	import java.util.LinkedList;
	import java.util.List;
	import java.util.Map;
	import java.util.Queue;
	import java.util.concurrent.ConcurrentLinkedQueue;

	import org.apache.catalina.comet.CometEvent;
	import org.json.JSONObject;
	import org.apache.cometd.bayeux.Bayeux;
	import org.apache.cometd.bayeux.Client;
	import org.apache.cometd.bayeux.Listener;
	import org.apache.cometd.bayeux.Message;
	import org.apache.juli.logging.Log;
	import org.apache.juli.logging.LogFactory;
	import java.util.concurrent.atomic.AtomicInteger;
	import java.util.HashMap;
	import java.util.ArrayList;

	public class ClientImpl implements Client {

	public static final int SUPPORT_CALLBACK_POLL = 0x1;
	public static final int SUPPORT_LONG_POLL = 0x2;

	public static final String COMET_EVENT_ATTR = "org.apache.cometd.bayeux.client";

	private static final Log log = LogFactory.getLog(ClientImpl.class);

	protected static LinkedList<Message> EMPTY_LIST = new LinkedList<Message>();
	/**
	* queued message for remote clients.
	*/
	protected LinkedList<Message> messages = null;

	/**
	*
	*/
	protected Queue<CometEvent> events = new LinkedList<CometEvent>();

	/**
	* Unique id representing this client
	*/
	protected String id;

	/**
	* supported connection types, defaults to long-polling
	*/
	protected int supportedConnTypes = SUPPORT_LONG_POLL \| SUPPORT_CALLBACK_POLL;

	/**
	* The desired connection type
	*/
	protected int desirectConnType = SUPPORT_LONG_POLL;

	/**
	* Does this client use json-comment-filtered messages
	*/
	protected boolean useJsonFiltered = false;

	/**
	* Same JVM clients, get local=true
	*/
	protected boolean local;

	/**
	* The callback object for local clients
	*/
	protected Listener listener;

	protected AtomicInteger nrofsubscriptions = new AtomicInteger(0);

	protected ClientImpl(String id, boolean local) {
	this.id = id;
	this.local = local;
	if (!local) messages = new LinkedList<Message>();
	}

	protected ClientImpl(String id, CometEvent event) {
	this(id,false);
	events = new ConcurrentLinkedQueue<CometEvent>();
	addCometEvent(event);
	}

	public synchronized void deliver(Message message) {
	deliverInternal(null,new MessageImpl[] {(MessageImpl)message});
	}

	public synchronized void deliver(Message[] message) {
	deliverInternal(null,message);
	}

	protected synchronized void deliverInternal(ChannelImpl channel, MessageImpl message) {
	deliverInternal(channel,new MessageImpl[] {message});
	}

	protected synchronized void deliverInternal(ChannelImpl channel, Message[] msgs) {
	if (isLocal()) {
	//local clients must have a listener
	ArrayList<Message> list = new ArrayList<Message>();
	for (int i=0; msgs!=null && i<msgs.length; i++) {
	//dont deliver to ourselves
	if (this!=msgs[i].getClient()) list.add(msgs[i]);
	}
	if (getListener() != null && list.size()>0) {
	getListener().deliver(list.toArray(new Message[0]));
	}
	} else {
	for (int i=0; msgs!=null && i<msgs.length; i++) {
	MessageImpl message = (MessageImpl)msgs[i];
	if (this==message.getClient()) {
	//dont deliver to ourself
	continue;
	}
	//we are not implementing forever responses, if the client is connected
	//then we will fire off the message
	//first we check to see if we have any existing connections we can piggy back on
	CometEvent event = events.poll();
	boolean delivered = false;
	//TODO TODO - check on thread safety, for writing and for getting last request.
	if (event!=null) {
	synchronized (event) {
	RequestBase rq = (RequestBase)event.getHttpServletRequest().getAttribute(RequestBase.LAST_REQ_ATTR);
	if (rq!=null) {
	Map map = new HashMap();
	try {
	map.put(Bayeux.CHANNEL_FIELD,message.getChannel().getId());
	map.put(Bayeux.DATA_FIELD,message);
	JSONObject json = new JSONObject(map);
	if (log.isDebugEnabled()) {
	log.debug("Message instantly delivered to remote client["+this+"] message:"+json);
	}
	rq.addToDeliveryQueue(this, json);
	//deliver the batch
	if (i==(msgs.length-1)) {
	rq.deliver(event, this);
	event.close(); //todo, figure out a better way, this means only one message gets delivered
	removeCometEvent(event); //and delivered instantly
	}
	delivered = true;
	} catch (Exception x) {
	log.error(x);
	}
	}
	}
	}
	if (!delivered) {
	if (log.isDebugEnabled()) {
	log.debug("Message added to queue for remote client["+this+"] message:"+message);
	}
	//queue the message for the next round
	messages.add(message);
	}
	}
	}
	}

	public String getId() {
	return this.id;
	}

	protected Listener getListener() {
	return listener;
	}

	public boolean hasMessages() {
	if (isLocal()) return false;
	else {
	return messages.size() > 0;
	}
	}

	public boolean isLocal() {
	return local;
	}

	public int getSupportedConnTypes() {
	return supportedConnTypes;
	}

	public int getDesirectConnType() {
	return desirectConnType;
	}

	public boolean useJsonFiltered() {
	return useJsonFiltered;
	}

	public void setListener(Listener listener) {
	this.listener = listener;
	}

	public void setSupportedConnTypes(int supportedConnTypes) {
	this.supportedConnTypes = supportedConnTypes;
	}

	public void setUseJsonFiltered(boolean useJsonFiltered) {
	this.useJsonFiltered = useJsonFiltered;
	}

	public void setDesirectConnType(int desirectConnType) {
	this.desirectConnType = desirectConnType;
	}

	public boolean supportsCallbackPoll() {
	return (supportedConnTypes & SUPPORT_CALLBACK_POLL) == SUPPORT_CALLBACK_POLL;
	}

	public boolean supportsLongPoll() {
	return (supportedConnTypes & SUPPORT_LONG_POLL) == SUPPORT_LONG_POLL;
	}

	public synchronized List<Message> takeMessages() {
	if (isLocal()) return null;
	if (messages.size()==0) return EMPTY_LIST;
	List result = new LinkedList(messages);
	messages.clear();
	return result;
	}

	public String toString() {
	StringBuilder buf = new StringBuilder(super.toString());
	buf.append(" id=").append(getId());
	return buf.toString();
	}

	public boolean isSubscribed() {
	return nrofsubscriptions.get()>0;
	}

	protected synchronized boolean addCometEvent(CometEvent event) {
	boolean result = false;
	if (!events.contains(event)) {
	events.add(event);
	result = true;
	}
	event.getHttpServletRequest().setAttribute(COMET_EVENT_ATTR,this);
	return result;
	}

	protected synchronized boolean removeCometEvent(CometEvent event) {
	boolean result = events.remove(event);
	event.getHttpServletRequest().removeAttribute(COMET_EVENT_ATTR);
	return result;
	}


	protected void subscribed(ChannelImpl ch) {
	nrofsubscriptions.addAndGet(1);
	}

	protected void unsubscribed(ChannelImpl ch) {
	nrofsubscriptions.addAndGet(-1);
	}

	public void startBatch(){
	//noop until improved
	}
	public void endBatch() {
	//noop until improved
	}

	}