/*
    * Replica is published under the terms
    * of the Apache Software License.
    */
   package replica.group.jgroups;
   
   import java.io.Serializable;
   import java.util.Iterator;
   import java.util.List;
  import java.util.Vector;
  import java.util.logging.Logger;
  
  import org.jgroups.Channel;
  import org.jgroups.ChannelClosedException;
  import org.jgroups.ChannelException;
  import org.jgroups.ChannelFactory;
  import org.jgroups.ChannelNotConnectedException;
  import org.jgroups.MembershipListener;
  import org.jgroups.Message;
  import org.jgroups.MessageListener;
  import org.jgroups.View;
  import org.jgroups.blocks.PullPushAdapter;
  import org.jgroups.stack.IpAddress;
  
  import replica.group.Address;
  import replica.group.CommunicationException;
  import replica.group.GroupManager;
  import replica.group.event.GroupListener;
  import replica.group.event.MessageEvent;
  import replica.group.event.SuspectEvent;
  import replica.group.event.ViewEvent;
  
  /***
   * Implementation of a group communications services provider using the
   * jgroups.org code.
   * 
   * @author Pedro Costa
   * @author Helder Silva
   * @since 19/Jan/2004
   */
  public class JGroupsGroupManager implements GroupManager {
  	
  	static Logger logger = Logger.getLogger(JGroupsGroupManager.class.getName());
  	
  	String groupName;
  	ChannelFactory channelFactory;
  	String channelProperties;
  	Channel channel;
  	
  	List groupListeneres = new Vector();
  	
  	PullPushAdapter block;
  
      /***
       * Default constructor.
       */
      public JGroupsGroupManager() {
          super();
      }
  
      /***
       * Adds a new listener for Group Events.
       * @see replica.group.GroupManager#addListener(replica.group.event.GroupListener)
       */
      public void addListener(GroupListener listener) {
      	logger.fine("Adding new GroupListener.");
      	
      	groupListeneres.add( listener );
      }
  
      /***
       * Return the name of the group under management.
       * 
       * @see replica.group.GroupManager#getGroupName()
       */
      public String getGroupName() {
          return groupName;
      }
  
      /***
       * Return a list of the current group members addresses.
       * 
       * @see replica.group.GroupManager#getGroupView()
       */
      public List getGroupView() {
      	
      	if( getChannel() == null )
  			throw new IllegalStateException("You are not connected to any group.");
  			
  		View view = getChannel().getView();
  		
  		return convertViewListFromJGroups( view );
      }
      
      /***
       * Return the local address
       * 
       * @see {@link GroupManager#getLocalManager()}
       */
     public Address getLocalAddress(){
     	return convertJGroupsIpAddressToIpAddress( (IpAddress)getChannel().getLocalAddress() );
     }
 
 	/***
 	 * Convert a JGroups View into a non specific implementation view.
 	 * 
 	 */
 	protected List convertViewListFromJGroups(View view) {
     	
 		List res = new Vector();
 			
 		if( view == null )
 			return res;
 			
 		List jgroupsList = view.getMembers();
 		
 		for (Iterator iter = jgroupsList.iterator(); iter.hasNext();) {
 			replica.group.IpAddress addr = 
 				convertJGroupsIpAddressToIpAddress( (IpAddress)iter.next() );
             
 			res.add(addr);
 		}
 		
 		return res;
 	}
 	
     /***
      * Join a group. You must set the channel and the group name before
      * calling this method.
      * 
      * @see replica.group.GroupManager#joinGroup()
      * @throws IllegalStateException if the channel is null or the group name
      * 									was not set.
      * @throws CommunicationException if there is a problem connecting to the
      * 				channel. You can look on the root cause of this exception
      * 				for further information.
      */
     public void joinGroup() {
     	
     	if( getChannel() == null || getGroupName() == null )
     		throw new IllegalStateException(
 				"You must set the channel and the group name before trying to join."
 			);
     		
     	if( getChannel().isConnected() )
     		getChannel().disconnect();
     	try{
 	    	getChannel().connect( getGroupName() );
     	}
     	catch(ChannelClosedException e){
     		throw new CommunicationException("The channel was closed. Look on the root"+
     		"cause for more information.", e);
     	}
     	catch(ChannelException e){
 			throw new CommunicationException("Problem connecting to channel. Look on the"+
 			" root cause for more information.", e);
     	}
     	
 		block = new PullPushAdapter( getChannel() );
 		
 		block.setListener( new MyMessageListener() );
 		
 		block.addMembershipListener( new MyMembershipListener() );
 		
 		block.start();
 		
 		logger.info("Joined group [" + getGroupName() + "]. Local address [" + getChannel().getLocalAddress() 
 			+ "]. View: {"+getChannel().getView()+"}.");
     }
 
     /***
      * Leave a group.
      * 
      * @see replica.group.GroupManager#leaveGroup()
      */
     public void leaveGroup() {
     	
     	if( getChannel() != null && getChannel().isConnected() )
     		getChannel().disconnect();
     		
 		logger.info("Leaved group [" + getGroupName() + "].");
     }
 
     /***
      * @see replica.group.GroupManager#sendMessage(replica.group.Address, java.lang.Object)
      */
     public void sendMessage(Address rAddr, Object message) {
     	
     	if( getChannel() == null )
     		throw new IllegalStateException("You are not connected to any group.");
     	
 		IpAddress jgroupsRecAddr = null;
 		if( rAddr != null )
 			jgroupsRecAddr = 
 				convertIpAddressToJGroupsIpAddress(
 					(replica.group.IpAddress)rAddr);
     	try{
 			logger.info("Sending message {" + message + "}.");
 			
 	    	getChannel().send(null, jgroupsRecAddr,	(Serializable)message);
     	}
     	catch(ChannelClosedException e){
 			throw new CommunicationException("Channel is closed. Look on the root"+
 						" cause for more information.",e);
     	}
 		catch(ChannelNotConnectedException e){
 			throw new CommunicationException("Channel not connected. Look on the root"+
 						" cause for more information.",e);
 		}
     }
     
     /***
      * Convert an address in non specific implementation to a jgroups 
      * class.
      * 
      * @param address
      * @return
      */
 	org.jgroups.stack.IpAddress convertIpAddressToJGroupsIpAddress(
 			replica.group.IpAddress address){
 		// TODO using the same class name is confusing
 		//    should revise this.
 		return new org.jgroups.stack.IpAddress( 
 						address.getAddress(),
 						address.getPort() );
 	}
 	
 	/***
 	 * Convert an address in jgroups class to an non specific implementation
 	 * class.
 	 * 
 	 * @param address
 	 * @return
 	 */
 	replica.group.IpAddress convertJGroupsIpAddressToIpAddress(
 		org.jgroups.stack.IpAddress address){
 		// TODO using the same class name is confusing
 		//    should revise this.
 		replica.group.IpAddress repAddr = new replica.group.IpAddress();
 		repAddr.setAddress( address.getIpAddress() );
 		repAddr.setPort( address.getPort() );
 		
 		return repAddr;
 	}
 
     /***
      * Short cut method. simply calls #sendMessage(Address, Object) with a null
      * address.
      * 
      * @see replica.group.GroupManager#sendMessage(java.lang.Object)
      */
     public void sendMessage(Object message) {
     	sendMessage(null, message);
     }
 
     /***
      * Sets the name of the group to be managed. You still have to call 
      * joinGroup after setting it's name.
      * 
      * @see replica.group.GroupManager#setGroupName(java.lang.String)
      */
     public void setGroupName(String groupName) {
     	this.groupName = groupName;
     }
 
     /***
      * @see replica.group.GroupManager#removeListener(replica.group.event.GroupListener)
      */
     public boolean removeListener(GroupListener listener) {
     	return groupListeneres.remove(listener);
     }
     
     /*
      *  ==== GroupListener notify methods ====
      */
      
     /***
      * Notifies GroupListener members that a new message was received.
      * 
      */
     void notifyMessage(Message message){
     	
     	MessageEvent event = new MessageEvent(
     			this, message.getObject(), 
     			convertJGroupsIpAddressToIpAddress( 
     					(IpAddress)message.getSrc() ) );
     					
 		for (Iterator iter = groupListeneres.iterator(); iter.hasNext();) {
 			((GroupListener) iter.next()).message( event);
 		}
     }
     
     /***
      * Notifies GroupListener members that view has changed.
      * 
      * @param view
      */
     void notifyViewChanged(View view){
     	logger.info("View: {" + view + "}.");
     	
     	ViewEvent event = new ViewEvent(this,
     		convertViewListFromJGroups( view ) );
     	
 		for (Iterator iter = groupListeneres.iterator(); iter.hasNext();) {
 			((GroupListener) iter.next()).viewChanged( event);
 		}
     }
     
     /***
      * Notifies GroupListener members that a suspect message was received.
      * 
      * @param address
      */
     void notifySuspect(org.jgroups.Address address){
     	
     	SuspectEvent event = new SuspectEvent( this,
     		convertJGroupsIpAddressToIpAddress((IpAddress)address) );
     		
 		for (Iterator iter = groupListeneres.iterator(); iter.hasNext();) {
 			((GroupListener) iter.next()).suspect( event);
 		}
     }
     
 	/***
 	 * My message listener implementation.
 	 * 
 	 * @author no119431
 	 *
 	 * To change the template for this generated type comment go to
 	 * Window>Preferences>Java>Code Generation>Code and Comments
 	 */
 	class MyMessageListener implements MessageListener{
 		
         /* (non-Javadoc)
          * @see org.jgroups.MessageListener#getState()
          */
         public byte[] getState() {
         	// I'm not interested in this.
             return null;
         }
 
         /***
          * Simply calls the notify method on the main class.
          * 
          * @see org.jgroups.MessageListener#receive(org.jgroups.Message)
          */
         public void receive(Message message) {
         	notifyMessage(message);
         }
 
         /* (non-Javadoc)
          * @see org.jgroups.MessageListener#setState(byte[])
          */
         public void setState(byte[] arg0) {
 			// I'm not interested in this.
         }
 
 	}
 
 	/***
 	 * My Membership Listener implementation.
 	 * 
 	 * @author no119431
 	 *
 	 * To change the template for this generated type comment go to
 	 * Window>Preferences>Java>Code Generation>Code and Comments
 	 */	
 	class MyMembershipListener implements MembershipListener{
 		
 	    /* (non-Javadoc)
          * @see org.jgroups.MembershipListener#block()
          */
         public void block() {
             // TODO Should i do something here?
 
         }
 
         /***
          * Simply calls the notify method on the main class.
          * 
          * @see org.jgroups.MembershipListener#suspect(org.jgroups.Address)
          */
         public void suspect(org.jgroups.Address address) {
         	
         	notifySuspect( address);
         }
 
         /***
          * Simply calls the notify method on the main class.
          * 
          * @see org.jgroups.MembershipListener#viewAccepted(org.jgroups.View)
          */
         public void viewAccepted(View view) {
         	
         	notifyViewChanged( view);
         }
 
 }
     /***
      * @return
      */
     public ChannelFactory getChannelFactory() {
         return channelFactory;
     }
 
     /***
      * @param factory
      */
     public void setChannelFactory(ChannelFactory factory) {
         channelFactory = factory;
     }
 
     /***
      * @return
      */
     public String getChannelProperties() {
         return channelProperties;
     }
 
     /***
      * @param string
      */
     public void setChannelProperties(String string) {
         channelProperties = string;
     }
     
     public Channel getChannel(){
     	
     	if( channel == null ){
     		try{
     			setChannel( 
     				getChannelFactory().createChannel( getChannelProperties() ) );
     		}
     		catch(ChannelException e){
     			throw new CommunicationException(e);
     		}
     	}
     		
     	return channel;
     }
     
     public void setChannel(Channel newChannel){
     	channel = newChannel;
     }
 
 	/***
 	 * Returns a copy of the list of current group listeners.
 	 * 
 	 * Mostly used for testing.
 	 * @return
 	 */
 	public List getGroupListeneres() {
 		return new Vector( groupListeneres );
 	}
 
 }