/**
    *
    */
   package org.sourceforge.jemm.comm.shared;
   
   import java.io.BufferedInputStream;
   import java.io.BufferedOutputStream;
   import java.io.IOException;
   import java.io.NotSerializableException;
  import java.io.ObjectInputStream;
  import java.io.ObjectOutputStream;
  import java.lang.reflect.InvocationTargetException;
  import java.lang.reflect.Method;
  import java.lang.reflect.Proxy;
  import java.util.ArrayList;
  import java.util.HashMap;
  import java.util.Map;
  import java.util.concurrent.ConcurrentHashMap;
  import java.util.concurrent.CountDownLatch;
  import java.util.concurrent.ExecutorService;
  import java.util.concurrent.SynchronousQueue;
  
  import org.apache.log4j.Logger;
  import org.sourceforge.jemm.comm.connection.Connection;
  import org.sourceforge.jemm.comm.connection.ConnectionException;
  
  /**
   * RPCHandler is used to implement the shared logic used by both {@link org.sourceforge.jemm.comm.server.RPCClient} 
   * and {@link org.sourceforge.jemm.comm.server.RPCServer}.
   * Given a socket RPCHandler basically manages the client and server interfaces, the proxies and the
   * actual message passing.
   *
   *  N.B. A client/server flag is passed to the constructor because {@link ObjectInputStream}/{@link ObjectOutputStream}
   *  must be invoked in alternate order on the two sides of the connection or it deadlocks.
   * @author Rory Graves
   *
   */
  public class RPCHandler implements Runnable {
      private static final Logger LOG = Logger.getLogger(RPCHandler.class);
      
      private static ThreadLocal<Object> connectionIdTL = new ThreadLocal<Object>();
  
      protected Connection connection;
      
      protected CountDownLatch initialisationLatch = new CountDownLatch(1);
  
      protected volatile boolean closing;
  
      protected ObjectInputStream is;
  
      protected ObjectOutputStream os;
  
      // interfaces offered on the remote side
      protected Map<Class<?>, Object> remoteInterfaces;
  
      // interfaces we offer to the rest of the world
      protected Map<Class<?>, Object> localInterfaces;
  
      protected ExecutorService requestExecutor;
  
      protected ThreadLocal<String> threadIdTL = new ThreadLocal<String>();
  
      protected ThreadLocal<SynchronousQueue<Message>> syncQueueTL = new ThreadLocal<SynchronousQueue<Message>>();
  
      protected ConcurrentHashMap<String, SynchronousQueue<Message>> msgSyncPoints = 
      	new ConcurrentHashMap<String, SynchronousQueue<Message>>();
  
      protected final boolean isClient;
  
      protected RPCHandlerListener listener = null;
  
      protected Object connectionId;
  
      /**
       * Create an RPCHandler
       * @param isClient Is this the client side of the connection (false for server)
       * @param connection The socket connection.
       * @param localInterfaces The interfaces this side offers.
       * @param requestExecutor The service to use for running requests.
       * @param connectionId The id of the connection.
       */
      public RPCHandler(boolean isClient,Connection connection, Map<Class<?>, Object> localInterfaces,
              ExecutorService requestExecutor,Object connectionId) {
          this.isClient = isClient;
          this.connection = connection;
          this.closing = false;
  
          this.localInterfaces = localInterfaces;
          this.requestExecutor = requestExecutor;
          this.connectionId = connectionId;
      }
  
      /**
       * Set the listener for this handler for event notification.
       * @param newListener The listener to inform of events.
       */
      public synchronized void setHandlerListener(RPCHandlerListener newListener) {
          this.listener = newListener;
      }
 
     /**
      * Returns an local interface for calling the remote interface.
      * @param ifClass The interface to get.
      * @return An object which offers the interface.
      * @throws IllegalArgumentException If the requested class is not an interface or not supported.
      */
     public synchronized Object getRemoteIF(Class<?> ifClass) {
         try {
             initialisationLatch.await();
         } catch (InterruptedException e) {
             // do nothing
         }
         
         if (!ifClass.isInterface())
             throw new IllegalArgumentException(
                     "given class is not an interface");
 
         Object obj = remoteInterfaces.get(ifClass);
         if (obj == null) {
             if (!remoteInterfaces.keySet().contains(ifClass))
                 throw new IllegalArgumentException("Interface " + ifClass
                         + " not offered by server");
 
             obj = createProxyClass(ifClass);
             remoteInterfaces.put(ifClass, obj);
         }
 
         return obj;
     }
 
     protected Object createProxyClass(Class<?> ifClass) {
         Class<?>[] ifs = {ifClass};
         RPCProxyHandler ph = new RPCProxyHandler(this, ifClass);
         Object obj = Proxy.newProxyInstance(this.getClass().getClassLoader(),
                 ifs, ph);
         return obj;
     }
 
     public void initialise() {
         try {
         	// n.b. order of initialisation is reversed from client to server as otherwise
         	// object streams fail to initialise
         	
             if(isClient) {
                 is = new ObjectInputStream(new BufferedInputStream(connection.getInputStream()));
                 os = new ObjectOutputStream(new BufferedOutputStream(connection.getOutputStream()));
                 
                 // this is needed otherwise buffering means that ObjectOutputStream fails to initialise
                 os.flush();  
             } else {
                 os = new ObjectOutputStream(new BufferedOutputStream(connection.getOutputStream()));
                 
                 // this is needed otherwise buffering means that ObjectOutputStream fails to initialise
                 os.flush();  
                 is = new ObjectInputStream(new BufferedInputStream(connection.getInputStream()));
             }
 
             AvailableIFsMessage ifOMsg = new AvailableIFsMessage(
                     localInterfaces.keySet().toArray(new Class<?>[0]));
             os.writeObject(ifOMsg);
             os.flush();
 
             AvailableIFsMessage ifIMsg = (AvailableIFsMessage) is .readObject();
 
             synchronized(this) {
             	remoteInterfaces = new HashMap<Class<?>, Object>();
             	for (Class<?> ifClass : ifIMsg.getOfferedIFs())
             		remoteInterfaces.put(ifClass, null);
             }
             
             if(isClient) {
             	// read the ready message
             	is.readObject();
             }
 
         } catch (ClassNotFoundException e) {
         	LOG.error("Caught exception whilst initialising ",e);
             connectionTerminated();
         } catch (IOException ioe) {
         	LOG.error("Caught exception whilst initialising ",ioe);
             connectionTerminated();
         }
     }
     
     public void run() {
         try {
             initialisationLatch.countDown();
             if(!isClient) {
             	os.writeObject(new ServerReadyMessage());
             	os.flush();
             }
                         
             while (true) {
                 Object o = null;
                 try {
                     o = is.readObject();
                 } catch (ClassNotFoundException e1) {
                     e1.printStackTrace();
                 }
 
                 if (o instanceof Message) {
                     Message message = (Message) o;
                     try {
                         receiveMessage(message);
                     } catch (Exception e) {
                         LOG.warn("exception thrown whilst sending message to receiver",e);
                     }
                 } else
                     LOG.warn("Invalid object on stream " + o);
             }
         } catch (IOException se) {
             connectionTerminated();
         }
     }
 
     protected void receiveMessage(Message message) {
         if (message instanceof RPCCallRespMessage) {
             final String threadId = message.getThreadId();
             SynchronousQueue<Message> replyQueue = msgSyncPoints.get(threadId);
             if (replyQueue != null) {
                 try {
                     replyQueue.put(message);
                 } catch (InterruptedException e) {
                     LOG.info("Receive thread interrupted");
                 }
             } else {
                 LOG.error("No client thread found for sync message to "
                         + threadId);
             }
         } else if (message instanceof RPCCallMessage) {
             processCallMessage((RPCCallMessage) message);
         } else if (message instanceof ErrorMessage) {
             LOG.warn("Error message recieved from server "
                     + ((ErrorMessage) message).getUserErrorMessage());
         } else
             LOG.warn("Invalid message type received by client: "
                     + message.getClass());
     }
 
     /**
      * Internal method to handle RPC call method.
      * This will invoke a thread using the executor supplied in the constructor.
      * @param message The message to process.
      */
 
     protected void processCallMessage(final RPCCallMessage message) {
     	if(closing) {
     		LOG.warn("Message processing skipped - client is shutting down");
     		return;
     	}
     	
         requestExecutor.execute(new Runnable() {
             public void run() {
                 connectionIdTL.set(connectionId);
                 try {
                     Class<?> targetIF  = message.getIfClass();
                     Object targetIFImpl = localInterfaces.get(message.getIfClass());
                     if(targetIF != null) {
                         Method method = targetIF.getMethod(message.getMethodName(), message.getParameterTypes());
                         if(method == null)
                             throw new IllegalArgumentException("Interface method does not exist");
                         Object resp = method.invoke(targetIFImpl,message.getParameters());
                         if(!message.isAsyncCall())
                             writeMessage(new RPCCallRespMessage(message.threadId,true,resp));
                     } else
                         throw new IllegalArgumentException("Interface not supported");
                 } catch(Exception e) {
                     Throwable cause = e instanceof InvocationTargetException?e.getCause():e;
                     if(!message.isAsyncCall())
                         writeMessage(new RPCCallRespMessage(message.threadId,false,cause));
                     else
                         LOG.warn("Exception caught whilst processing async call to " 
                         		+ message.getIfClass() + "." + message.getMethodName() + "()",e);
                 }
                 connectionIdTL.set(null);
             }
         });
     }
 
     protected synchronized void connectionTerminated() {
         if (!closing) {
             if(isClient)
                 LOG.error("Lost connection to Server");
             else
                 LOG.info("Lost connection to client");
         }
 
         closing = true;
         if (msgSyncPoints.size() > 0) {
             LOG.error("Client connection closed with waiters active");
             ArrayList<String> list = new ArrayList<String>();
             list.addAll(msgSyncPoints.keySet());
 
             ErrorMessage errorMessage = new ErrorMessage(
                     "Server connection lost");
             for (String threadId : list) {
                 SynchronousQueue<Message> queue = msgSyncPoints.get(threadId);
                 if (queue != null && !queue.offer(errorMessage))
                 	LOG.warn("Unable to inform thread " + threadId
                 			+ " of connection close");
             }
         }
 
         if(listener != null)
             listener.connectionTerminated();
     }
 
 
     /**
      * Close the connection to the server.
      */
     public void close() {
         closing = true;
         try {
         	connection.close();
         } catch (final ConnectionException ce) {
             LOG.warn("Exception thrown whilst closing client connection", ce);
         }
     }
 
     protected synchronized void writeMessage(Message message) {
         try {
             os.writeObject(message);
             os.flush();
         } catch(NotSerializableException nse) {
             LOG.error("Sent message not serializable ",nse);
         } catch (final IOException e) {
             LOG.warn("error caught writing object", e);
         }
     }
 
     /**
      * Send a synchronous message to the server. This method sends the given message 
      * and waits for a response message.
      *
      * @param message The message to send
      * @return The message received in reply
      */
     public Message sendSyncMessage(Message message) {
         final String threadId = ThreadUtil.getThreadId();
 
         SynchronousQueue<Message> sq = syncQueueTL.get();
         if (sq == null) {
             sq = new SynchronousQueue<Message>();
             syncQueueTL.set(sq);
         }
 
         msgSyncPoints.put(threadId, sq);
 
         writeMessage(message);
         Message replyMsg = null;
         try {
             replyMsg = sq.take();
         } catch (InterruptedException ie) {
             replyMsg = new ErrorMessage(
                     "InterruptedException received whilst waiting for reply");
         }
 
         msgSyncPoints.remove(threadId);
 
         return replyMsg;
     }
 
 
     /**
      * Internal method to make an asynchronous call to a given method.
      * @param ifClass The interface class being called.
      * @param methodName The method being called.
      * @param parameterTypes The parameter types of the method
      * @param args The actual call arguments (based types wrapped by Proxy.invoke).
      */
     protected void makeAsyncCall(Class<?> ifClass, String methodName,Class<?>[] parameterTypes, Object[] args) {
         RPCCallMessage callMessage = new RPCCallMessage(ThreadUtil.getThreadId(),true,
         		ifClass,methodName,parameterTypes,args);
         writeMessage(callMessage);
     }
 
     /**
      * Internal method to make a make a synchronous call to a given method.
      * @param ifClass The interface class being called.
      * @param methodName The method being called.
      * @param parameterTypes The parameter types of the method
      * @param args The actual call arguments (based types wrapped by Proxy.invoke).
      * @return The response message received  from the target.
      */
     protected RPCCallRespMessage makeSyncCall(Class<?> ifClass, String methodName,Class<?>[] parameterTypes,
             Object[] args) {
         RPCCallMessage callMessage = new RPCCallMessage(ThreadUtil.getThreadId(),false,
         		ifClass,methodName,parameterTypes,args);
         Message replyMsg = sendSyncMessage(callMessage);
         if(replyMsg instanceof RPCCallRespMessage)
             return (RPCCallRespMessage) replyMsg;
         else if(replyMsg instanceof ErrorMessage)
             return new RPCCallRespMessage(ThreadUtil.getThreadId(),false,
             		new IllegalStateException("Error message returned:" 
             				+ ((ErrorMessage) replyMsg).getUserErrorMessage()));
             
         else
             return new RPCCallRespMessage(ThreadUtil.getThreadId(),false,
             		new IllegalStateException("Unexpected message returned " + replyMsg.getClass()));
     }
 
     /**
      * Retrieve the connection id object associated with the current processing thread.
      * A connection id is associated with each execution by the thread-pool.
      * @return The connection id object supplied.
      */
     public static Object getConnectionId() {
         return connectionIdTL.get();
     }
 
     /**
      * Start the RPCHandler.  This will trigger the handler to initialise
      * the connection and be ready to start serving calls.
      */
     public void start() {
         (new Thread(this)).start();
     }
 }