Start line:  
End line:  

Snippet Preview

Snippet HTML Code

Stack Overflow Questions
  /*
   * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS HEADER.
   *
   * Copyright (c) 2007-2010 Oracle and/or its affiliates. All rights reserved.
   *
   * The contents of this file are subject to the terms of either the GNU
   * General Public License Version 2 only ("GPL") or the Common Development
   * and Distribution License("CDDL") (collectively, the "License").  You
   * may not use this file except in compliance with the License.  You can
  * obtain a copy of the License at
  * https://glassfish.dev.java.net/public/CDDL+GPL_1_1.html
  * or packager/legal/LICENSE.txt.  See the License for the specific
  * language governing permissions and limitations under the License.
  *
  * When distributing the software, include this License Header Notice in each
  * file and include the License file at packager/legal/LICENSE.txt.
  *
  * GPL Classpath Exception:
  * Oracle designates this particular file as subject to the "Classpath"
  * exception as provided by Oracle in the GPL Version 2 section of the License
  * file that accompanied this code.
  *
  * Modifications:
  * If applicable, add the following below the License Header, with the fields
  * enclosed by brackets [] replaced by your own identifying information:
  * "Portions Copyright [year] [name of copyright owner]"
  *
  * Contributor(s):
  * If you wish your version of this file to be governed by only the CDDL or
  * only the GPL Version 2, indicate your decision by adding "[Contributor]
  * elects to include this software in this distribution under the [CDDL or GPL
  * Version 2] license."  If you don't indicate a single choice of license, a
  * recipient has the option to distribute your version of this file under
  * either the CDDL, the GPL Version 2 or to extend the choice of license to
  * its licensees as provided above.  However, if you add GPL Version 2 code
  * and therefore, elected the GPL Version 2 license, then the option applies
  * only if the new code is made subject to such option by the copyright
  * holder.
  */
 
 package com.sun.grizzly.cometd;
 
This class implement the Bayeux Server side protocol.

Author(s):
Jeanfrancois Arcand
TAKAI, Naoto
 
 public class BayeuxCometHandler extends BayeuxCometHandlerBase{
     
     public final static String DEFAULT_CONTENT_TYPE ="application/json-comment-filtered";
             
     public final static String BAYEUX_COMET_HANDLER = "bayeuxCometHandler";
 
         = Collections.unmodifiableCollection(Collections.singleton());
 
             = new ConcurrentHashMap<String,Collection<String>>();
     
     private Queue<StringauthenticatedUsers 
             = new ConcurrentQueue<String>("BayeuxCometHandler.authenticatedUsers");
     
     private Random random = new Random();
     
     
             new ConcurrentHashMap<String,DataHandler>();
 
     
    public void onHandshake(CometEvent eventthrows IOException {            
        CometdContext cometdContext = (CometdContext)event.attachment();
        
        CometdRequest req = cometdContext.getRequest();
        CometdResponse res = cometdContext.getResponse();
        HandshakeRequest handshakeReq = (HandshakeRequest)cometdContext.getVerb();
        HandshakeResponse handshakeRes = new HandshakeResponse(handshakeReq);
        handshakeRes.setAdvice(new Advice());
        if (handshakeReq.isValid()) {
            String clientId  = null;
            synchronized(){
                clientId = String.valueOf(Long.toHexString(.nextLong()));
            }
            handshakeRes.setClientId(clientId);
            .offer(clientId);
        } else {
            handshakeRes.setSuccessful(false);
            handshakeRes.setError("501::invalid handshake");
        }
        res.setContentType();            
        res.write(handshakeRes.toJSON());
        res.flush();
    }
    
    
    public void onConnect(CometEvent eventthrows IOException {
        CometdContext cometdContext = (CometdContext)event.attachment();
        
        CometdRequest req = cometdContext.getRequest();
        CometdResponse res = cometdContext.getResponse();
        ConnectRequest connectReq = (ConnectRequest)cometdContext.getVerb();
        
        ConnectResponse connectRes = new ConnectResponse(connectReq);
        connectRes.setAdvice(new Advice());
        
        String errorMessage = isAuthenticatedAndValid(connectReq);      
        res.setContentType();            
        if (errorMessage == null){
            boolean hasChannel = false;
            String clientId = connectReq.getClientId();
            DataHandler dataHandler = .get(clientId);
            if (dataHandler != null && dataHandler.getChannels().size() > 0) {
                hasChannel = true;
                dataHandler.attach(new Object[]{req,res});
                event.getCometContext().addCometHandler(dataHandler);  
                connectRes.setAdvice(null);
            }
            String jsonMessage = (hasChannel ? 
                connectRes.toLongPolledJSON() : connectRes.toJSON());
            res.write(jsonMessage);
        } else {
            res.write(errorMessage);
        }
        res.flush();
    }
    
    
    @SuppressWarnings("unchecked")
    public void onDisconnect(CometEvent eventthrows IOException {
        CometdContext cometdContext = (CometdContext)event.attachment();
        
        CometdResponse res = cometdContext.getResponse();
        DisconnectRequest disconnectReq = (DisconnectRequest)cometdContext.getVerb();
        DisconnectResponse disconnectRes = new DisconnectResponse(disconnectReq);
        
        String errorMessage = isAuthenticatedAndValid(disconnectReq);
        res.setContentType();    
        if (errorMessage == null){
            removeActiveHandler(disconnectReq.getClientId(),event.getCometContext());
            .remove(disconnectReq.getClientId());
            res.write(disconnectRes.toJSON());
        } else {
            res.write(errorMessage);
        }
        res.flush();
        notifyEnd(eventdisconnectReq);
    }
    
    
    
Remove an DataHandler from the list of active listener.

Parameters:
clientId
ctx
Returns:
Return true if the clientId was a registered DataHandler
    public boolean removeActiveHandler(String clientIdCometContext ctx
            throws IOException{
        
        DataHandler dataHandler = 
                .remove(clientId);
        if (dataHandler != null && ctx.isActive(dataHandler)){
            ctx.notify("disconnecting",
                     .,dataHandler.hashCode());
            return true;
        }       
        return false;
    }
       
    public void onReconnect(CometEvent eventthrows IOException {
        CometdContext cometdContext = (CometdContext)event.attachment();
        
        CometdRequest req = cometdContext.getRequest();
        CometdResponse res = cometdContext.getResponse();
        ReconnectRequest reconnectReq = (ReconnectRequest)cometdContext.getVerb();                    
        ReconnectResponse reconnectRes = new ReconnectResponse(reconnectReq);
        
        String errorMessage = isAuthenticatedAndValid(reconnectReq);
        res.setContentType();            
        if (errorMessage == null){
            res.write(reconnectRes.toJSON());
        } else {
            res.write(errorMessage);
        }
        res.flush();
        notifyEnd(eventreconnectReq);
    }
    
    
    @SuppressWarnings("unchecked")
    public void onSubscribe(CometEvent eventthrows IOException {
        CometdContext cometdContext = (CometdContext)event.attachment();
        
        CometdRequest req = cometdContext.getRequest();
        CometdResponse res = cometdContext.getResponse();
        SubscribeRequest subscribeReq = (SubscribeRequest)cometdContext.getVerb();   
        SubscribeResponse subscribeRes = new SubscribeResponse(subscribeReq);
        
        String errorMessage = isAuthenticatedAndValid(subscribeReq);    
        res.setContentType();             
        if (errorMessage == null){
            String clientId = subscribeReq.getClientId();
            DataHandler dataHandler = .get(clientId);
            if (dataHandler == null) {
                dataHandler = new DataHandler(this);
                dataHandler.setClientId(clientId);
                .putIfAbsent(clientIddataHandler);
                // just in case two threads both put
                dataHandler = .get(clientId);
            }
            dataHandler.addChannel(subscribeReq.getSubscription());
            res.write(subscribeRes.toJSON());
        } else {
            res.write(errorMessage);
        }            
        res.flush();
        notifyEnd(eventsubscribeReq);
    }
    
    
    @SuppressWarnings("unchecked")
    public void onUnsubscribe(CometEvent eventthrows IOException {
        CometdContext cometdContext = (CometdContext)event.attachment();
        
        CometdRequest req = cometdContext.getRequest();
        CometdResponse res = cometdContext.getResponse();
        UnsubscribeRequest unsubscribeReq = (UnsubscribeRequest)cometdContext.getVerb();
        UnsubscribeResponse unsubscribeRes = new UnsubscribeResponse(unsubscribeReq);
        
        boolean hasSubscription = false;
        DataHandler dataHandler = null;
        String clientId = unsubscribeReq.getClientId();
        String subscription = unsubscribeReq.getSubscription();
        String errorMessage = isAuthenticatedAndValid(unsubscribeReq);       
        res.setContentType();  
        if (errorMessage == null){
            dataHandler = .get(clientId);
            if (dataHandler != null) {
                hasSubscription = dataHandler.removeChannel(subscription);
                if (hasSubscription) {
                    dataHandler.removeChannel(subscription);
                    Collection<StringunsubscribedChannels = .get(clientId);
                    if (unsubscribedChannels == null) {
                        unsubscribedChannels = new ConcurrentQueue<String>("BayeuxCometHandler.unsubscribedChannels");
                        .putIfAbsent(clientIdunsubscribedChannels);
                        // just in case two threads both put
                        unsubscribedChannels = .get(clientId);
                    }
                    unsubscribedChannels.add(subscription);
                }
            }
            unsubscribeRes.setSuccessful(hasSubscription);
            res.write(unsubscribeRes.toJSON());
        } else {
            res.write(errorMessage);
        }
        res.flush();
        notifyEnd(eventunsubscribeReq);
    }
    
    
    @SuppressWarnings("unchecked")
    public void onPublish(CometEvent eventthrows IOException {
        CometdContext cometdContext = (CometdContext)event.attachment();
        
        CometdResponse res = cometdContext.getResponse();
        PublishRequest publishReq = (PublishRequest)cometdContext.getVerb();
        PublishResponse publishRes = new PublishResponse(publishReq);
        DeliverResponse deliverRes = null;
                        
        String errorMessage = isAuthenticatedAndValid(publishReq);  
        res.setContentType();             
        if (errorMessage != null){
            res.write(errorMessage);
        } else {
            publishRes.setSuccessful(true);
            Data data = publishReq.getData();
            if (data != null) {
                deliverRes = new DeliverResponse(publishReq);
                deliverRes.setFollow(true);
                if (publishReq.isFirst()) {
                    deliverRes.setFirst(false);
                }
            }
         
            boolean hasWritten = false;
            if (deliverRes != null) {
                String clientId = publishReq.getClientId();
                if (clientId != null) {
                    Collection<StringsubscribedChannels = null;
                    DataHandler dataHandler = .get(clientId);
                    if (dataHandler != null) {
                        subscribedChannels = dataHandler.getChannels();
                    }
                    if (subscribedChannels != null &&
                            subscribedChannels.contains(publishReq.getChannel())) {
                        hasWritten = true;
                        if (publishReq.isLast()) {
                            publishRes.setLast(false);
                        }
                        res.write(publishRes.toJSON());
                        res.write(deliverRes.toJSON());
                    }
                }
            }
            if (!hasWritten) {
                res.write(publishRes.toJSON());
            }
        }
        res.flush(); 
        if (deliverRes != null) {
            event.getCometContext().notify(deliverRes);
        }
        notifyEnd(eventpublishReq);
    }
    private void notifyEnd(CometEvent eventVerbBase verbthrows IOException {
        if (verb.isLast()) {
            //XXX what if clientId is null
            String clientId = verb.getClientId();
            if (clientId != null) {
                Collection<StringsubscribedChannels = null;
                DataHandler dataHandler = .get(clientId);
                if (dataHandler != null) {
                    subscribedChannels = dataHandler.getChannels();
                }
                Collection<StringunsubscribedChannels = .get(clientId);
                if (subscribedChannels != null && subscribedChannels.size() > 0 ||
                        unsubscribedChannels != null && unsubscribedChannels.size() > 0) {
                    event.getCometContext().notify("NOTIFY_END");
                }
                if (unsubscribedChannels != null) {
                    unsubscribedChannels.clear();
                    .remove(clientId);
                }
            }
        }
    }
    
    
    
Has the client been authenticated (by executing the /meta/handshake operation) and the request valid.
    private String isAuthenticatedAndValid(VerbBase verb){
        String clientId = verb.getClientId();
        
        if (clientId == nullreturn null;
        if (clientId!= null && !.contains(clientId)){
            return constructError("402","Unknown Client"verb.getMetaChannel());
        } else if (!verb.isValid()){
            return constructError("501","Invalid Operation"verb.getMetaChannel());
        } else {
            return null;
        }
    }
    

    
Construct an error message.

Parameters:
errorMessage
errorMsg
meta
Returns:
    private final static String constructError(String errorMessage
            String errorMsg
            String meta){
        
        StringBuilder sb = new StringBuilder();
        sb.append("[{\"successful\":false,\"error\":\"");
        sb.append(errorMessage);
        sb.append("::");
        sb.append(errorMsg);
        sb.append("\",\"advice\":{\"reconnect\":\"handshake\"},\"channel\":\"");
        sb.append(meta);
        sb.append("\"}]");
        
        return sb.toString();
    }
    
    
    public final static CometdContext newCometdContext(final CometdRequest req
            final CometdResponse res,final Verb verb){
        return new CometdContext(){
            
            public Verb getVerb(){
                return verb;
            }
            
            public CometdRequest getRequest(){
                return req;
            }
                      
            public CometdResponse getResponse(){
                return res;
            }            
        };
    }  
    
        
    @Override
    public void onTerminate(CometEvent eventthrows IOException {
        onInterrupt(event);    
    }
    
    
    @Override
    public void onInterrupt(CometEvent eventthrows IOException {
    }
    
    // ---------------------------------------------------- Reserved but not used
    
    
    public void onPing(CometEvent eventthrows IOException {
    }
    
    
    public void onStatus(CometEvent eventthrows IOException {
    }
    
    public Collection<StringgetChannels() {
        return ;
    }
    
    public void addChannel(String channel) {
        // Not supported
    }
    public boolean removeChannel(String channel) {
        // Not supported
        return false;
    }
 
New to GrepCode? Check out our FAQ X