seti.java

是离开的肌肤了卡机是离开的就富利卡及是了的开发及拉考试及的福利科技阿斯利康的肌肤莱卡及时的离开福建阿斯顿发

package org.cometd.oort;

import java.util.Collection;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;

import org.cometd.Channel;
import org.cometd.Client;
import org.cometd.Message;
import org.cometd.MessageListener;
import org.mortbay.component.AbstractLifeCycle;
import org.mortbay.util.LazyList;
import org.mortbay.util.MultiMap;
import org.mortbay.util.ajax.JSON;
import org.mortbay.util.ajax.JSON.Output;

/* ------------------------------------------------------------ */
/** The Search for Extra Terrestial Intelligence.
 * 
 * Well in this case, just the search for a user logged onto an
 * Cometd node in an Oort cluster.
 * <p>
 * Seti allows an application to maintain a mapping from userId to 
 * comet client ID using the {@link #associate(String, Client)} and
 * {@link #disassociate(String)} methods. Each cometd node keeps its
 * own associate mapping for clients connected to it.
 * <p>
 * The {@link #sendMessage(Collection, String, Object)} and 
 * {@link #sendMessage(String, String, Object)} methods may be
 * used to send a message to user(s) anywhere in the Oort cluster
 * and Seti organizes the search of the distributed associate
 * maps in order to locate the user(s)
 * <p>
 * If users can be directed to shards of cometd servers, then
 * each Seti instance must be told it's shard ID and the {@link #userId2Shard(String)}
 * method must be extended to map users to shards.
 * 
 */
public class Seti extends AbstractLifeCycle
{
    public final static String SETI_ATTRIBUTE="org.cometd.oort.Seti";
    public final static String SETI_SHARD="seti.shard";
    
    final String _setiId;
    final String _setiChannelId;
    final String _shardId;
    final Oort _oort;
    final Client _client;
    final ShardLocation _allShardLocation;
    final Channel _setiIdChannel;
    final Channel _setiAllChannel;
    final Channel _setiShardChannel;
    
    final ConcurrentMap<String, Location> _uid2Location = new ConcurrentHashMap<String, Location>();
    
    /* ------------------------------------------------------------ */
    public Seti(Oort oort, String shardId)
    {
        _oort=oort;
        _client = _oort.getBayeux().newClient("seti");
        _setiId=_oort.getURL().replace("://","_").replace("/","_").replace(":","_");
        _shardId=shardId;

        _setiChannelId="/seti/"+_setiId;
        _setiIdChannel=_oort.getBayeux().getChannel(_setiChannelId,true);
        _setiIdChannel.setPersistent(true);
        _oort.observeChannel(_setiIdChannel.getId());
        _setiIdChannel.subscribe(_client);
        
        _setiAllChannel=_oort.getBayeux().getChannel("/seti/ALL",true);
        _setiAllChannel.setPersistent(true);
        _oort.observeChannel(_setiAllChannel.getId());
        _setiAllChannel.subscribe(_client);
        
        _setiShardChannel=_oort.getBayeux().getChannel("/seti/"+shardId,true);
        _setiShardChannel.setPersistent(true);
        _oort.observeChannel(_setiShardChannel.getId());
        _setiShardChannel.subscribe(_client);
        
        _allShardLocation = new ShardLocation(_setiAllChannel);
         
    }

    /* ------------------------------------------------------------ */
    protected void doStart()
        throws Exception
    {
        super.doStart();
        _client.addListener(new MessageListener()
        {
            public void deliver(Client from, Client to, Message msg)
            {
                receive(from,to,msg);
            }
        });
    }
    
    /* ------------------------------------------------------------ */
    protected void doStop()
        throws Exception
    {
        _client.disconnect();
    }
    
    /* ------------------------------------------------------------ */
    public void associate(final String userId,final Client client)
    {
        _uid2Location.put(userId,new LocalLocation(client));
        userId2Shard(userId).associate(userId);
    }

    /* ------------------------------------------------------------ */
    public void disassociate(final String userId)
    {
        _uid2Location.remove(userId);
        userId2Shard(userId).disassociate(userId);
    }

    /* ------------------------------------------------------------ */
    public void sendMessage(final String toUser,final String toChannel,final Object message)
    {
        Location location = _uid2Location.get(toUser);
        if (location==null)
            location = userId2Shard(toUser);
        
        location.sendMessage(toUser,toChannel,message);
    }

    /* ------------------------------------------------------------ */
    public void sendMessage(final Collection<String> toUsers,final String toChannel,final Object message)
    {
        // break toUsers in to shards
        MultiMap shard2users = new MultiMap();
        for (String userId:toUsers)
        {       
            ShardLocation shard = userId2Shard(userId);
            shard2users.add(shard,userId);
        }
        
        // for each shard
        for (Map.Entry<ShardLocation,Object> entry : (Set<Map.Entry<ShardLocation,Object>>)shard2users.entrySet())
        {
            // TODO, we could look at all users in shard to see if we
            // know a setiId for each, and if so, break the user list
            // up into a message for each seti-id. BUT it is probably 
            // more efficient just to send to the entire shard (unless
            // the number of nodes in the shard is greater than the
            // number of users).
            
            ShardLocation shard = entry.getKey();
            Object lazyUsers = entry.getValue();
            
            if (LazyList.size(lazyUsers)==1)
                shard.sendMessage((String)lazyUsers,toChannel,message);
            else
                shard.sendMessage((List<String>)lazyUsers,toChannel,message);
        }
    }
    
    /* ------------------------------------------------------------ */
    protected ShardLocation userId2Shard(final String userId)
    {
        return _allShardLocation;
    }

    /* ------------------------------------------------------------ */
    protected void receive(final Client from, final Client to, final Message msg)
    {
        //System.err.println("SETI "+_oort+":: "+msg);

        if (!(msg.getData() instanceof Map))
            return;
        
        // extract the message details
        Map<String,Object> data = (Map<String,Object>)msg.getData();
        final String toUid=(String)data.get("to");
        final String fromUid=(String)data.get("from");
        final Object message = data.get("message");
        final String on = (String)data.get("on");
        
        // Handle any client locations contained in the message
        if (fromUid!=null)
        {
            if (on!=null)
            {
                //System.err.println(_oort+":: "+fromUid+" on "+on);
                _uid2Location.put(fromUid,new SetiLocation("/seti/"+on));
            }
            else 
            {
                final String off = (String)data.get("off");
                if (off!=null)
                {
                    //System.err.println(_oort+":: "+fromUid+" off ");
                    _uid2Location.remove(fromUid,new SetiLocation("/seti/"+off));
                }
            }
        }
        
        // deliver message
        if (message!=null && toUid!=null)
        {
            final String toChannel=(String)data.get("channel");
            Location location=_uid2Location.get(toUid);
            
            if (location==null && _setiChannelId.equals(msg.getChannel()))
                // was sent to this node, so escalate to the shard.
                location =userId2Shard(toUid);
            
            if (location!=null)
                location.receive(toUid,toChannel,message);
        }
        
    }


    /* ------------------------------------------------------------ */
    /* ------------------------------------------------------------ */
    private interface Location
    {
        public void sendMessage(String toUser,String toChannel,Object message);
        public void receive(String toUser,String toChannel,Object message);
    }
    

    /* ------------------------------------------------------------ */
    /* ------------------------------------------------------------ */
    class LocalLocation implements Location
    {
        Client _client;
        
        LocalLocation(Client client)
        {
            _client=client;
        }

        public void sendMessage(String toUser, String toChannel, Object message)
        {
            _client.deliver(Seti.this._client,toChannel,message,null);
        }

        public void receive(String toUser, String toChannel, Object message)
        {
            _client.deliver(Seti.this._client,toChannel,message,null);
        }
    }

    /* ------------------------------------------------------------ */
    /* ------------------------------------------------------------ */
    class SetiLocation implements Location
    {
        Channel _channel;

        SetiLocation(String channelId)
        {
            _channel=_oort._bayeux.getChannel(channelId,true);
        }
        
        SetiLocation(Channel channel)
        {
            _channel=channel;
        }
        
        public void sendMessage(String toUser, String toChannel, Object message)
        {
            _channel.publish(Seti.this._client,new SetiMessage(toUser,toChannel,message),null);
        }

        public void receive(String toUser, String toChannel, Object message)
        {
            
        }

        public boolean equals(Object o)
        {
            return o instanceof SetiLocation &&
            ((SetiLocation)o)._channel.equals(_channel);
        }
        
        public int hashCode()
        {
            return _channel.hashCode();
        }
    }

    /* ------------------------------------------------------------ */
    /* ------------------------------------------------------------ */
    class ShardLocation implements Location
    {
        Channel _channel;
        
        ShardLocation(String shardId)
        {
            _channel=_oort._bayeux.getChannel("/seti/"+shardId,true);
            
        }
        
        ShardLocation(Channel channel)
        {
            _channel=channel;
        }
        
        public void sendMessage(final Collection<String> toUsers, final String toChannel, final Object message)
        {
            _channel.publish(Seti.this._client,new SetiMessage(toUsers,toChannel,message),null);
        }

        public void sendMessage(String toUser, String toChannel, Object message)
        {
            _channel.publish(Seti.this._client,new SetiMessage(toUser,toChannel,message),null);
        }
        
        public void receive(String toUser, String toChannel, Object message)
        {
            
        }
        
        public void associate(final String user)
        {
            _channel.publish(Seti.this._client,new SetiPresence(user,true),null);
        }
        
        public void disassociate(final String user)
        {
            _channel.publish(Seti.this._client,new SetiPresence(user,false),null);
        }
    }

    /* ------------------------------------------------------------ */
    /* ------------------------------------------------------------ */
    class SetiMessage implements JSON.Convertible
    {
        String _toUser;
        Collection<String> _toUsers;
        String _toChannel;
        Object _message;

        SetiMessage(String toUser,String toChannel, Object message)
        {
            _toUser=toUser;
            _toChannel=toChannel;
            _message=message;
        }
        
        SetiMessage(Collection<String> toUsers,String toChannel, Object message)
        {
            _toUsers=toUsers;
            _toChannel=toChannel;
            _message=message;
        }
        
        public void fromJSON(Map object)
        {
            throw new UnsupportedOperationException();
        }

        public void toJSON(Output out)
        {
            if (_toUser!=null)
                out.add("to",_toUser);
            else if (_toUsers!=null)
                out.add("to",_toUsers);
            out.add("channel",_toChannel);
            out.add("from",_setiId);
            out.add("message",_message);
        }   
    }
    
    /* ------------------------------------------------------------ */
    /* ------------------------------------------------------------ */
    class SetiPresence implements JSON.Convertible
    {
        String _user;
        boolean _on;

        SetiPresence(String user,boolean on)
        {
            _user=user;
            _on=on;
        }
        
        public void fromJSON(Map object)
        {
            throw new UnsupportedOperationException();
        }

        public void toJSON(Output out)
        {
            out.add("from",_user);
            out.add(_on?"on":"off",_setiId);
        }
    }
    
}