// ========================================================================
// Copyright 2006 Mort Bay Consulting Pty. Ltd.
// ------------------------------------------------------------------------
// Licensed 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.mortbay.cometd;

import java.io.IOException;
import java.security.SecureRandom;
import java.util.HashMap;
import java.util.Map;
import java.util.Random;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;

import javax.servlet.ServletContext;

import org.mortbay.util.DateCache;
import org.mortbay.util.ajax.JSON;

import dojox.cometd.Bayeux;
import dojox.cometd.Channel;
import dojox.cometd.Client;
import dojox.cometd.DataFilter;
import dojox.cometd.SecurityPolicy;

/* ------------------------------------------------------------ */
/**
 * @author gregw
 * @author aabeling: added JSONP transport
 * 
 */
public abstract class AbstractBayeux implements Bayeux
{   
    public static final ChannelId META_ID=new ChannelId(META);
    public static final ChannelId META_CONNECT_ID=new ChannelId(META_CONNECT);
    public static final ChannelId META_CLIENT_ID=new ChannelId(META_CLIENT);
    public static final ChannelId META_DISCONNECT_ID=new ChannelId(META_DISCONNECT);
    public static final ChannelId META_HANDSHAKE_ID=new ChannelId(META_HANDSHAKE);
    public static final ChannelId META_PING_ID=new ChannelId(META_PING);
    public static final ChannelId META_RECONNECT_ID=new ChannelId(META_RECONNECT); // deprecated
    public static final ChannelId META_STATUS_ID=new ChannelId(META_STATUS);
    public static final ChannelId META_SUBSCRIBE_ID=new ChannelId(META_SUBSCRIBE);
    public static final ChannelId META_UNSUBSCRIBE_ID=new ChannelId(META_UNSUBSCRIBE);
    
    public static final JSON.Literal TRANSPORTS=new JSON.Literal("[\"long-polling\",\"callback-polling\"]");

    private static final JSON.Literal EXT_JSON_COMMENTED=new JSON.Literal("{\"json-comment-filtered\":true}");
    private static HashMap<String,Class> _transports=new HashMap<String,Class>();
    
    private static final JSON.Literal __NO_ADVICE=new JSON.Literal("{}");

    /*
    AtomicInteger _publish = new AtomicInteger();
    AtomicInteger _deliver = new AtomicInteger();
    
    {
        new Thread(new Runnable()
        {
            public void run()
            {
                try
                {
                    while(true)
                    {
                        Thread.sleep(1000);
                        System.err.println(_clients.size()+" "+_publish.get()+" "+_deliver.get());
                    }
                }
                catch (InterruptedException e)
                {
                    e.printStackTrace();
                }


            }
            
        }).start();
        
    }*/
    
    static
    {
        // _transports.put("iframe",IFrameTransport.class);
        _transports.put("long-polling",JSONTransport.class);
        _transports.put("callback-polling",JSONPTransport.class);
    }
    HashMap<String,Handler> _handlers=new HashMap<String,Handler>();
    
    ChannelImpl _root = new ChannelImpl("/",this);
    ConcurrentHashMap<String,ClientImpl> _clients=new ConcurrentHashMap<String,ClientImpl>();
    SecurityPolicy _securityPolicy=new DefaultPolicy();
    Object _advice=new JSON.Literal("{\"reconnect\":\"retry\",\"interval\":0}");
    int _adviceVersion=0;
    Object _unknownAdvice=new JSON.Literal("{\"reconnect\":\"handshake\",\"interval\":500}");
    int _logLevel;
    long _clientTimeoutMs=300000;
    boolean _JSONCommented;
    boolean _initialized;
    ConcurrentHashMap<String, Set<String>> _browserPolls=new ConcurrentHashMap<String, Set<String>>();
    int _multiFrameInterval=-1;
    JSON.Literal _multiFrameAdvice; 
    
    transient ServletContext _context;
    transient Random _random;
    transient DateCache _dateCache;
    transient ConcurrentHashMap<String, ChannelId> _channelIdCache;
    private boolean _alwaysResumePoll;

    /* ------------------------------------------------------------ */
    /**
     * @param context.
     *            The logLevel init parameter is used to set the logging to
     *            0=none, 1=info, 2=debug
     */
    protected AbstractBayeux()
    {
        _handlers.put("*",new PublishHandler());
        _handlers.put(META_HANDSHAKE,new HandshakeHandler());
        _handlers.put(META_CONNECT,new ConnectHandler());
        _handlers.put(META_RECONNECT,new ReConnectHandler());
        _handlers.put(META_DISCONNECT,new DisconnectHandler());
        _handlers.put(META_SUBSCRIBE,new SubscribeHandler());
        _handlers.put(META_UNSUBSCRIBE,new UnsubscribeHandler());
        _handlers.put(META_STATUS,new StatusHandler());
        _handlers.put(META_PING,new PingHandler());
    }

    /* ------------------------------------------------------------ */
    /**
     * @param channels
     *            A {@link ChannelId}
     * @param filter
     *            The filter instance to apply to new channels matching the
     *            pattern
     */
    public void addFilter(String channels, DataFilter filter)
    {
        synchronized (this)
        {
            ChannelImpl channel = (ChannelImpl)getChannel(channels,true);
            channel.addDataFilter(filter);
        }
    }

    /* ------------------------------------------------------------ */
    public void removeFilter(String channels, DataFilter filter)
    {
        synchronized (this)
        {
            ChannelImpl channel = (ChannelImpl)getChannel(channels,false);
            if (channel!=null)
                channel.removeDataFilter(filter);
        }
    }

    /* ------------------------------------------------------------ */
    /**
     * @param id
     * @return
     */
    public ChannelImpl getChannel(ChannelId id)
    {
        return _root.getChild(id);
    }

    /* ------------------------------------------------------------ */
    /* (non-Javadoc)
     * @see org.mortbay.cometd.Bx#getChannel(java.lang.String)
     */
    public ChannelImpl getChannel(String id)
    {
        ChannelId cid=getChannelId(id);
        if (cid.depth()==0)
            return null;
        return _root.getChild(cid);
    }

    /* ------------------------------------------------------------ */
    /* (non-Javadoc)
     * @see org.mortbay.cometd.Bx#getChannel(java.lang.String, boolean)
     */
    public Channel getChannel(String id, boolean create)
    {
        synchronized(this)
        {
            ChannelImpl channel=getChannel(id);

            if (channel==null && create)
            {
                channel=new ChannelImpl(id,this);
                _root.addChild(channel);
                
                if (isLogInfo())
                    logInfo("newChannel: "+channel);
            }
            return channel;
        }
    }
    
    /* ------------------------------------------------------------ */
    public ChannelId getChannelId(String id)
    {
        ChannelId cid = _channelIdCache.get(id);
        if (cid==null)
        {
            // TODO shrink cache!
            cid=new ChannelId(id);
            _channelIdCache.put(id,cid);
        }
        return cid;
    }

    /* ------------------------------------------------------------ */
    /* (non-Javadoc)
     * @see org.mortbay.cometd.Bx#getClient(java.lang.String)
     */
    public Client getClient(String client_id)
    {
        synchronized(this)
        {
            if (client_id==null)
                return null;
            Client client = _clients.get(client_id);
            return client;
        }
    }

    /* ------------------------------------------------------------ */
    /* (non-Javadoc)
     * @see org.mortbay.cometd.Bx#getClientIDs()
     */
    public Set getClientIDs()
    {
        return _clients.keySet();
    }

    /* ------------------------------------------------------------ */
    public long getClientTimeoutMs()
    {
        return _clientTimeoutMs;
    }

    /* ------------------------------------------------------------ */
    /**
     * @return the logLevel. 0=none, 1=info, 2=debug
     */
    public int getLogLevel()
    {
        return _logLevel;
    }

    /* ------------------------------------------------------------ */
    /* ------------------------------------------------------------ */
    /* (non-Javadoc)
     * @see org.mortbay.cometd.Bx#getSecurityPolicy()
     */
    public SecurityPolicy getSecurityPolicy()
    {
        return _securityPolicy;
    }
    
    /* ------------------------------------------------------------ */
    /** Handle a Bayeux message.
     * This is normally only called by the bayeux servlet or a test harness.
     * @param client The client if known
     * @param transport The transport to use for the message
     * @param message The bayeux message.
     */
    public String handle(ClientImpl client, Transport transport, Map<String,Object> message) throws IOException
    {
        String channel_id=(String)message.get(CHANNEL_FIELD);
        // System.err.println(channel_id+" : "+message);
        
        Handler handler=(Handler)_handlers.get(channel_id);
        if (handler==null)
            handler=(Handler)_handlers.get("*");

        if (isLogDebug())
            logDebug("handler="+handler);

        handler.handle(client,transport,message);
        
        return channel_id;
    }

    /* ------------------------------------------------------------ */
    public boolean hasChannel(String id)
    {
        ChannelId cid=getChannelId(id);
        return _root.getChild(cid)!=null;
    }

    /* ------------------------------------------------------------ */
    public boolean isInitialized()
    {
        return _initialized;
    }

    /* ------------------------------------------------------------ */
    /**
     * @return the commented
     */
    public boolean isJSONCommented()
    {
        return _JSONCommented;
    }

    /* ------------------------------------------------------------ */
    public boolean isLogDebug()
    {
        return _logLevel>1;
    }


    /* ------------------------------------------------------------ */
    public boolean isLogInfo()
    {
        return _logLevel>0;
    }
    
    /* ------------------------------------------------------------ */
    public void logDebug(String message)
    {
        if (_logLevel>1)
            _context.log(message);
    }

    /* ------------------------------------------------------------ */
    public void logDebug(String message, Throwable th)
    {
        if (_logLevel>1)
            _context.log(message,th);
    }

    /* ------------------------------------------------------------ */
    public void logWarn(String message, Throwable th)
    {
        _context.log(message+": "+th.toString());
    }

    /* ------------------------------------------------------------ */
    public void logWarn(String message)
    {
        _context.log(message);
    }

    /* ------------------------------------------------------------ */
    public void logInfo(String message)
    {
        if (_logLevel>0)
            _context.log(message);
    }
    
    /* ------------------------------------------------------------ */
    public Client newClient(String idPrefix,dojox.cometd.Listener listener)
    {
        return new ClientImpl(this,idPrefix,listener);
    }

    /* ------------------------------------------------------------ */
    public abstract ClientImpl newRemoteClient();

    /* ------------------------------------------------------------ */
    /** Create new transport object for a bayeux message
     * @param client The client
     * @param message the bayeux message
     * @return the negotiated transport.
     */
    public Transport newTransport(ClientImpl client, Map message)
    {
        if (isLogDebug())
            logDebug("newTransport: client="+client+",message="+message);

        Transport result=null;

        try
        {
            String type=client==null?null:client.getConnectionType();
            if (type==null)
                type=(String)message.get("connectionType");

            /* check if jsonp parameter is set */
            if (type==null)
            {
                String jsonp=(String)message.get("jsonp");
                if (jsonp!=null)
                {
                    /* use JSONPTransport */
                    if (isLogDebug())
                        logDebug("newTransport: using JSONPTransport with jsonp="+jsonp);
                    result=new JSONPTransport(client!=null&&client.isJSONCommented());
                    ((JSONPTransport)result).setJsonp(jsonp);
                }
            }

            if ((type!=null)&&(result==null))
            {
                Class trans_class=(Class)_transports.get(type);
                if (trans_class!=null)
                {
                    if (trans_class.equals(JSONPTransport.class))
                    {
                        String jsonp=(String)message.get("jsonp");
                        if (jsonp==null)
                        {
                            throw new Exception("JSONPTransport needs jsonp parameter");
                        }
                        result=new JSONPTransport(client!=null&&client.isJSONCommented());
                        ((JSONPTransport)result).setJsonp(jsonp);
                    }
                    else
                    {
                        result=(Transport)(trans_class.newInstance());
                        result.setJSONCommented(client!=null&&client.isJSONCommented());
                    }
                }
            }

            if (result==null)
            {
                result=new JSONTransport(client!=null&&client.isJSONCommented());
            }
        }
        catch (Exception e)
        {
            throw new RuntimeException(e);
        }

        if (isLogDebug())
            logDebug("newTransport: result="+result);
        return result;
    }

    /* ------------------------------------------------------------ */
    /* (non-Javadoc)
     * @see org.mortbay.cometd.Bx#publish(org.mortbay.cometd.ChannelId, org.mortbay.cometd.Client, java.lang.Object, java.lang.String)
     */
    public void publish(ChannelId to, Client from, Object data, String msgId)
    {
        HashMap<String,Object> msg = new Message();
        msg.put(CHANNEL_FIELD,to.toString());
        msg.put(TIMESTAMP_FIELD,getTimeOnServer());
        if (msgId==null)
        {
            long id=msg.hashCode()
            ^(to==null?0:to.hashCode())
            ^(from==null?0:from.hashCode());
            id=id<0?-id:id;
            msg.put(ID_FIELD,Long.toString(id,36));
        }
        else
            msg.put(ID_FIELD,msgId);
            
        msg.put(DATA_FIELD,data);
        _root.deliver(to,from,msg);
    }

    /* ------------------------------------------------------------ */
    public void publish(Client fromClient, String toChannelId, Object data, String msgId)
    {
        publish(getChannelId(toChannelId),fromClient,data,msgId);
    }


    /* ------------------------------------------------------------ */
    /* (non-Javadoc)
     * @see dojox.cometd.Bayeux#deliver(dojox.cometd.Client, java.lang.String, java.util.Map)
     */
    public void deliver(Client fromClient,Client toClient, String toChannel, Map<String, Object> message)
    {
        if (toChannel!=null)
            message.put(Bayeux.CHANNEL_FIELD,toChannel);
        if (!message.containsKey(Bayeux.TIMESTAMP_FIELD))
            message.put(Bayeux.TIMESTAMP_FIELD,getTimeOnServer());
        if (toClient!=null)
            toClient.deliver(fromClient,message);
    }
    

    /* ------------------------------------------------------------ */
    public boolean removeChannel(ChannelId channelId)
    {
        // TODO Auto-generated method stub
        return false;
    }

    /* ------------------------------------------------------------ */
    /* (non-Javadoc)
     * @see org.mortbay.cometd.Bx#removeClient(java.lang.String)
     */
    Client removeClient(String client_id)
    {
        ClientImpl client;
        synchronized(this)
        {
            if (client_id==null)
                return null;
            client = _clients.remove(client_id);
        }
        if (client!=null)
        {
            client.unsubscribeAll();
        }
        return client;
    }

    /* ------------------------------------------------------------ */
    public void setClientTimeoutMs(long ms)
    {
        _clientTimeoutMs=ms;
    }

    /* ------------------------------------------------------------ */
    /**
     * @param commented the commented to set
     */
    public void setJSONCommented(boolean commented)
    {
        _JSONCommented=commented;
    }

    /* ------------------------------------------------------------ */
    /**
     * @param logLevel
     *            the logLevel: 0=none, 1=info, 2=debug
     */
    public void setLogLevel(int logLevel)
    {
        _logLevel=logLevel;
    }
    
    /* ------------------------------------------------------------ */
    /* (non-Javadoc)
     * @see org.mortbay.cometd.Bx#setSecurityPolicy(org.mortbay.cometd.SecurityPolicy)
     */
    public void setSecurityPolicy(SecurityPolicy securityPolicy)
    {
        _securityPolicy=securityPolicy;
    }

    /* ------------------------------------------------------------ */
    public void subscribe(String toChannel, Client subscriber)
    {
        ChannelImpl channel = (ChannelImpl)getChannel(toChannel,true);
        if (channel!=null)
            channel.subscribe(subscriber);
    }

    /* ------------------------------------------------------------ */
    public void unsubscribe(String toChannel, Client subscriber)
    {
        ChannelImpl channel = (ChannelImpl)getChannel(toChannel);
        if (channel!=null)
            channel.unsubscribe(subscriber);
    }

    
    /* ------------------------------------------------------------ */
    /**
     * @return the multiFrameInterval in milliseconds
     */
    public int getMultiFrameInterval()
    {
        return _multiFrameInterval;
    }

    /* ------------------------------------------------------------ */
    /**
     * The time a client should delay between reconnects when multiple
     * connections from the same browser are detected. This effectively 
     * produces traditional polling.
     * @param multiFrameInterval the multiFrameInterval to set
     */
    public void setMultiFrameInterval(int multiFrameInterval)
    {
        _multiFrameInterval=multiFrameInterval;
        if (multiFrameInterval>0)
            _multiFrameAdvice=new JSON.Literal("{\"reconnect\":\"retry\",\"interval\":"+_multiFrameInterval+",\"multiple-clients\":true}");
        else
            _multiFrameAdvice=new JSON.Literal("{\"reconnect\":\"none\",\"interval\":30000,\"multiple-clients\":true}");
    }


    /* ------------------------------------------------------------ */
    public Object getAdvice()
    {
        return _advice;
    }


    /* ------------------------------------------------------------ */
    public void setAdvice(Object advice)
    {
        synchronized(this)
        {
            _advice=advice;
            _adviceVersion++;
        }
    }
    
    /* ------------------------------------------------------------ */
    /**
     * 
     */
    protected void initialize(ServletContext context)
    {
        synchronized(this)
        {
            _initialized=true;
            _context=context;
            try
            {
                _random=SecureRandom.getInstance("SHA1PRNG");
            }
            catch (Exception e)
            {
                context.log("Could not get secure random for ID generation",e);
                _random=new Random();
            }
            _random.setSeed(_random.nextLong()^hashCode()^(context.hashCode()<<32)^Runtime.getRuntime().freeMemory());
            _dateCache=new DateCache();
            _channelIdCache=new ConcurrentHashMap<String, ChannelId>();
            
            _root.addChild(new ServiceChannel("/service"));
            
        }
    }


    /* ------------------------------------------------------------ */
    public long getRandom(long variation)
    {
        long l=_random.nextLong()^variation;
        return l<0?-l:l;
    }

    /* ------------------------------------------------------------ */
    /**
     * @return
     */
    public String getTimeOnServer()
    {
        return _dateCache.format(System.currentTimeMillis());
    }

    /* ------------------------------------------------------------ */
    /* ------------------------------------------------------------ */
    public static class DefaultPolicy implements SecurityPolicy
    {
        public boolean canHandshake(Map message)
        {
            return true;
        }
        
        public boolean canCreate(Client client, String channel, Map message)
        {
            return client!=null && !channel.startsWith("/meta/");
        }

        public boolean canSubscribe(Client client, String channel, Map message)
        {
            return client!=null && !channel.startsWith("/meta/");
        }

        public boolean canPublish(Client client, String channel, Map message)
        {
            return client!=null && !channel.startsWith("/meta/");
        }

    }


    /* ------------------------------------------------------------ */
    /* ------------------------------------------------------------ */
    private abstract class Handler
    {
        abstract void handle(ClientImpl client, Transport transport, Map<String, Object> message) throws IOException;
        
        void unknownClient(Transport transport,String channel) throws IOException
        {
            Map<String,Object> reply=new HashMap<String,Object>();
            reply.put(CHANNEL_FIELD,channel);
            reply.put(SUCCESSFUL_FIELD,Boolean.FALSE);
            reply.put(ERROR_FIELD,"402::Unknown client");
            reply.put("advice",new JSON.Literal("{\"reconnect\":\"handshake\"}"));
            transport.send(reply);
        }
    }

    /* ------------------------------------------------------------ */
    /* ------------------------------------------------------------ */
    private class ConnectHandler extends Handler
    {
        protected String _metaChannel=META_CONNECT;
        
        public void handle(ClientImpl client, Transport transport, Map<String, Object> message) throws IOException
        {      
            if (client==null)
            {
                unknownClient(transport,_metaChannel);
                return;
            }

            // is this the first connect message?
            String type=client.getConnectionType();
            boolean polling=true;
            if (type==null)
            {
                type=(String)message.get("connectionType");
                client.setConnectionType(type);
                polling=transport.keepAlive();
            }
            
            Object advice=null; 
        
            if (polling && _multiFrameInterval>0)
            {
                String browser_id=client.getBrowserId();
                if (browser_id!=null)
                {
                    int count=client.onBrowser(browser_id);
                    if (count>1)
                    {
                        polling=false;
                        advice=_multiFrameAdvice;
                        client.setAdviceVersion(-1);
                    }
                }
            }

            synchronized(this)
            {
                if (client.getAdviceVersion()!=_adviceVersion && client.getAdviceVersion()>=0)
                {
                    advice=_advice;
                    client.setAdviceVersion(_adviceVersion);
                }
            }
           
            // reply to connect message
            String id=(String)message.get(ID_FIELD); 
            JSON.Literal reply = new JSON.Literal(
                    "{\""+
                    CHANNEL_FIELD+"\":\""+META_CONNECT+"\",\""+
                    SUCCESSFUL_FIELD+"\":"+Boolean.TRUE+",\""+
                    TIMESTAMP_FIELD+"\":\""+getTimeOnServer()+"\""+
                    (advice!=null?(",\""+ADVICE_FIELD+"\":"+JSON.toString(advice)):"")+
                    (id!=null?(",\""+ID_FIELD+"\":"+JSON.toString(id)+"}"):"}"));
            
            transport.send(reply);
            transport.setPolling(polling);
        }
    }
    
    /* ------------------------------------------------------------ */
    /* ------------------------------------------------------------ */
    private class ReConnectHandler extends ConnectHandler
    {
        {
            _metaChannel=META_RECONNECT;
        }
    }
    
    /* ------------------------------------------------------------ */
    /* ------------------------------------------------------------ */
    private class DisconnectHandler extends Handler
    {
        public void handle(ClientImpl client, Transport transport, Map<String, Object> message) throws IOException
        {
            if (client==null)
            {
                unknownClient(transport,META_DISCONNECT);
                return;
            }
                
            client.remove(false);
            
            Map<String,Object> reply=new HashMap<String,Object>();
            reply.put(CHANNEL_FIELD,META_DISCONNECT);
            reply.put(SUCCESSFUL_FIELD,Boolean.TRUE);
            reply.put(TIMESTAMP_FIELD,getTimeOnServer());
            Object id=message.get(ID_FIELD); 
            if (id!=null)
                reply.put(ID_FIELD,id.toString());
            
            transport.send(reply);
            transport.setPolling(false);
        }
    }


    /* ------------------------------------------------------------ */
    /* ------------------------------------------------------------ */
    private class HandshakeHandler extends Handler
    {
        public void handle(ClientImpl client, Transport transport, Map<String, Object> message) throws IOException
        {
            if (client!=null)
                throw new IllegalStateException();

            if (_securityPolicy!=null && !_securityPolicy.canHandshake(message))
            {
                Map<String,Object> reply=new HashMap<String,Object>();
                reply.put(CHANNEL_FIELD,META_HANDSHAKE);
                reply.put(SUCCESSFUL_FIELD,Boolean.FALSE);
                reply.put(ERROR_FIELD,"403::Handshake denied");
                reply.put("advice",new JSON.Literal("{\"reconnect\":\"handshake\",\"interval\":1000}"));
                transport.send(reply);
                return;
            }
            
            client=newRemoteClient();

            Map ext = (Map)message.get(EXT_FIELD);

            boolean commented=_JSONCommented && ext!=null && ((Boolean)ext.get("json-comment-filtered")).booleanValue();
            
            Map<String,Object> reply=new HashMap<String,Object>();
            reply.put(CHANNEL_FIELD,META_HANDSHAKE);
            reply.put("version","1.0");
            reply.put("minimumVersion","0.9");
            if (isJSONCommented())
                reply.put(EXT_FIELD,EXT_JSON_COMMENTED);

            if (client!=null)
            {
                reply.put("supportedConnectionTypes",TRANSPORTS);
                reply.put("successful",Boolean.TRUE);
                reply.put(CLIENT_FIELD,client.getId());
                if (_advice!=null)
                    reply.put(ADVICE_FIELD,_advice);
                client.setJSONCommented(commented);
                transport.setJSONCommented(commented);
            }
            else
            {
                reply.put("successful",Boolean.FALSE);
                if (_advice!=null)
                    reply.put(ADVICE_FIELD,_advice);
            }

            if (isLogDebug())
                logDebug("handshake.handle: reply="+reply);

            Object id=message.get(ID_FIELD);
            if (id!=null)
                reply.put(ID_FIELD,id.toString());
            
            transport.send(reply);
        }
    }

    /* ------------------------------------------------------------ */
    /* ------------------------------------------------------------ */
    private class PingHandler extends Handler
    {
        public void handle(ClientImpl client, Transport transport, Map<String, Object> message) throws IOException
        {
        }
    }

    /* ------------------------------------------------------------ */
    /* ------------------------------------------------------------ */
    private class PublishHandler extends Handler
    {
        public void handle(ClientImpl client, Transport transport, Map<String, Object> message) throws IOException
        {
            String channel_id=(String)message.get(CHANNEL_FIELD);
            
            if (client==null && message.containsKey(CLIENT_FIELD))
            {
                unknownClient(transport,channel_id);
                return;
            }
            
            Object id=message.get(ID_FIELD);

            ChannelId cid=getChannelId(channel_id);
            Object data=message.get("data");
                
            if (data!=null&&_securityPolicy.canPublish(client,channel_id,message))
            {
                JSON.Literal reply = new JSON.Literal(
                        "{\""+
                        SUCCESSFUL_FIELD+"\":"+Boolean.TRUE+",\""+
                        CHANNEL_FIELD+"\":"+JSON.toString(channel_id)+",\""+
                        TIMESTAMP_FIELD+"\":\""+getTimeOnServer()+"\""+
                        (id!=null?(",\""+ID_FIELD+"\":"+JSON.toString(id.toString())+"}"):"}"));

                transport.send(reply);
                publish(cid,client,data,id==null?null:id.toString());
            }
            else
            {
                JSON.Literal reply = new JSON.Literal(
                        "{\""+
                        SUCCESSFUL_FIELD+"\":"+Boolean.FALSE+",\""+
                        ERROR_FIELD+"\":\"404::Unknown Channel\",\""+
                        CHANNEL_FIELD+"\":"+JSON.toString(channel_id)+",\""+
                        TIMESTAMP_FIELD+"\":\""+getTimeOnServer()+"\""+
                        (id!=null?(",\""+ID_FIELD+"\":"+JSON.toString(id.toString())+"}"):"}"));

                transport.send(reply);
            }
        }
    }

    /* ------------------------------------------------------------ */
    /* ------------------------------------------------------------ */
    private class StatusHandler extends Handler
    {
        public void handle(ClientImpl client, Transport transport, Map<String, Object> message) throws IOException
        {
        }
    }

    /* ------------------------------------------------------------ */
    /* ------------------------------------------------------------ */
    private class SubscribeHandler extends Handler
    {
        public void handle(ClientImpl client, Transport transport, Map<String, Object> message) throws IOException
        {
            if (client==null)
            {
                unknownClient(transport,META_SUBSCRIBE);
                return;
            }

            String subscribe_id=(String)message.get(SUBSCRIPTION_FIELD);

            // select a random channel ID if none specifified
            if (subscribe_id==null)
            {
                subscribe_id=Long.toString(getRandom(message.hashCode()^client.hashCode()),36);
                while (getChannel(subscribe_id)!=null)
                    subscribe_id=Long.toString(getRandom(message.hashCode()^client.hashCode()),36);
            }

            ChannelId cid=null;
            boolean can_subscribe=false;
            boolean service=false;
            
            if (subscribe_id.startsWith(Bayeux.SERVICE_SLASH))
            {
                can_subscribe=true;
                service=false;
            }
            else if (subscribe_id.startsWith(Bayeux.META_SLASH))
            {
                can_subscribe=false;
            }
            else
            {
                cid=getChannelId(subscribe_id);
                can_subscribe=_securityPolicy.canSubscribe(client,subscribe_id,message);
            }
                
            Map<String,Object> reply=new HashMap<String,Object>();
            reply.put(CHANNEL_FIELD,META_SUBSCRIBE);
            reply.put(SUBSCRIPTION_FIELD,subscribe_id);

            if (can_subscribe)
            {
                if (cid!=null)
                {
                    ChannelImpl channel=getChannel(cid);
                    if (channel==null&&_securityPolicy.canCreate(client,subscribe_id,message))
                        channel=(ChannelImpl)getChannel(subscribe_id, true);

                    if (channel!=null)
                        channel.subscribe(client);
                    else
                        can_subscribe=false;
                }
                        
                if (can_subscribe)
                {
                    reply.put(SUCCESSFUL_FIELD,Boolean.TRUE);
                }
                else 
                {
                    reply.put(SUCCESSFUL_FIELD,Boolean.FALSE);
                    reply.put(ERROR_FIELD,"403::cannot create");
                }
            }
            else
            {
                reply.put(SUCCESSFUL_FIELD,Boolean.FALSE);
                reply.put(ERROR_FIELD,"403::cannot subscribe");
                
            }

            Object id=message.get(ID_FIELD); 
            if (id!=null)
                reply.put(ID_FIELD,id.toString());
            transport.send(reply);
        }
    }

    /* ------------------------------------------------------------ */
    /* ------------------------------------------------------------ */
    private class UnsubscribeHandler extends Handler
    {
        public void handle(ClientImpl client, Transport transport, Map<String, Object> message) throws IOException
        {
            if (client==null)
            {
                unknownClient(transport,META_UNSUBSCRIBE);
                return;
            }

            String channel_id=(String)message.get(SUBSCRIPTION_FIELD);
            ChannelImpl channel=getChannel(channel_id);
            if (channel!=null)
                channel.unsubscribe(client);

            Map<String,Object> reply=new HashMap<String,Object>();
            reply.put(CHANNEL_FIELD,META_UNSUBSCRIBE);
            if (channel!=null)
            {
                channel.unsubscribe(client);
                reply.put(SUBSCRIPTION_FIELD,channel.getId());
                reply.put(SUCCESSFUL_FIELD,Boolean.TRUE);
            }
            else
                reply.put(SUCCESSFUL_FIELD,Boolean.FALSE);
                
            Object id=message.get(ID_FIELD); 
            if (id!=null)
                reply.put(ID_FIELD,id.toString());
            transport.send(reply);
        }
    }

    /* ------------------------------------------------------------ */
    /* ------------------------------------------------------------ */
    private class ServiceChannel extends ChannelImpl
    {
        ServiceChannel(String id)
        {
            super(id,AbstractBayeux.this);
        }

        /* ------------------------------------------------------------ */
        /* (non-Javadoc)
         * @see org.mortbay.cometd.ChannelImpl#addChild(org.mortbay.cometd.ChannelImpl)
         */
        public void addChild(ChannelImpl channel)
        {
            super.addChild(channel);
            setPersistent(true);
        }

        /* ------------------------------------------------------------ */
        public void subscribe(Client client)
        {
            if (client.isLocal())
                super.subscribe(client);
        }
        
    }

}
