/*
      Copyright 2008 Ramon Servadei
   
      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 fulmine.rpc;
  
  import static fulmine.util.Utils.COMMA;
  import static fulmine.util.Utils.EMPTY_STRING;
  import static fulmine.util.Utils.UNDERSCORE;
  import static fulmine.util.Utils.safeToString;
  import static fulmine.util.Utils.string;
  
  import java.lang.reflect.Method;
  import java.util.Arrays;
  import java.util.Iterator;
  import java.util.Map;
  import java.util.Set;
  import java.util.Map.Entry;
  
  import org.apache.commons.logging.Log;
  
  import fulmine.AbstractLifeCycle;
  import fulmine.Domain;
  import fulmine.Type;
  import fulmine.context.IFrameworkContext;
  import fulmine.event.EventFrameExecution;
  import fulmine.model.container.IContainer;
  import fulmine.model.field.FieldUtils;
  import fulmine.model.field.IField;
  import fulmine.model.field.IntegerField;
  import fulmine.model.field.StringField;
  import fulmine.util.Utils;
  import fulmine.util.collection.CollectionFactory;
  import fulmine.util.log.AsyncLog;
  import fulmine.util.reference.is;
  
  /**
   * Standard {@link IRpcRegistry} implementation. This has a single
   * {@link IRecord} that represents the set of RPC definitions that are published
   * by local application code. Remote contexts will subscribe for this to receive
   * the current and future RPC definitions. The record is addressed using the
   * following attributes:
   * <ol>
   * <li>identity={@link IRpcRegistry#RPC_REGISTRY}
   * <li>type={@link Type#RECORD}
   * <li>domain={@link Domain#FRAMEWORK}
   * </ol>
   * When a new RPC definition is published by application code, a new field is
   * added to the registry record. The field is called the RPC registry key. The
   * RPC registry key format is described in the following ABNF:
   * 
   * <pre>
   * RPC key              =  &quot;rpc&quot; counter &quot;_&quot; compactForm
   * counter              =  1*DIGIT ; this increments each time a new definition is registered
   * compactForm          =  ALPHA 0*(ALPHA/DIGIT) argTypeCompactForm; this is the RPC name
   * argTypeCompactForm   =  0*ALPHA ; this is a string formed from the first letter of each argument type
   * </pre>
   * 
   * This ensures that the key is deterministic. Application code can cycle
   * through the fields of the RPC registry record and intercept any beginning
   * with this pattern in order to find RPC definitions. The value of the key is
   * the string representation of the RPC definition. E.g. rpc12lasersIDB is the
   * key for the 12th registered RPC, the RPC definition is lasers(IntegerField,
   * DoubleField, BooleanField) - compact form is lasersIDB
   * <p>
   * An {@link IRpcDefinition} can re-construct itself from this representation so
   * this is what is sent on-the-wire to remote contexts. After the key is added,
   * an update event is then triggered for the registry which will be received by
   * all subscribers for the registry (including channels connected to remote
   * contexts).
   * <p>
   * When an RPC definition is unpublished, the key and value are removed.
   * <p>
   * This class is thread safe.
   * 
   * @author Ramon Servadei
   */
  public final class RpcRegistry extends AbstractLifeCycle implements
      IRpcRegistry
  {
      /** The local context */
      private final IFrameworkContext context;
  
      /**
       * The registry for all RPC definitions published by the local context.
       */
      private IContainer registry;
  
     /**
      * The registry key count. Uses the <a
      * href="http://www.ibm.com/developerworks/java/library/j-jtp06197.html"
      * >'cheap read-write lock'</a>
      */
     volatile int keyCount;
 
     /** Tracks the published RPC definitions against their RPC registry key */
     private final Map<String, IRpcDefinition> published;
 
     /** Maps the RPC registry key to the handler for the RPC */
     private final Map<String, IRpcHandler> handlers;
 
     /**
      * Construct the RPC repository
      * 
      * @param context
      *            the local context
      */
     public RpcRegistry(IFrameworkContext context)
     {
         super();
         this.context = context;
         this.published = CollectionFactory.newMap();
         this.handlers = CollectionFactory.newMap();
     }
 
     public boolean publishProdedure(IRpcHandler handler,
         IRpcDefinition rpcDefinition)
     {
         final String rpcDefAsString = rpcDefinition.toString();
         try
         {
             synchronized (getPublished())
             {
                 if (getPublished().values().contains(rpcDefinition))
                 {
                     return false;
                 }
                 // temporary addition
                 getPublished().put(rpcDefAsString, rpcDefinition);
             }
 
             // lock the record
             getRegistry().beginFrame(new EventFrameExecution());
             try
             {
                 final String[] fields = getRegistry().getComponentIdentities();
 
                 /*
                  * Find the key for the new RPC definition. We need to go
                  * through all current keys to see if any can be re-used because
                  * an unpublish will just set the string field value to null.
                  */
                 String keyName = null;
                 for (String fieldName : fields)
                 {
                     final IField field = getRegistry().get(fieldName);
                     if (field instanceof StringField)
                     {
                         final StringField key = (StringField) field;
                         if (key.get() == null)
                         {
                             keyName = fieldName;
                             key.set(rpcDefAsString);
                             break;
                         }
                     }
                 }
                 if (keyName == null)
                 {
                     final IntegerField rpcCount =
                         getRegistry().<IntegerField> get(
                             IRpcRegistry.RPC_KEY_COUNT);
                     keyName =
                         IRpcRegistry.RPC_KEY + this.keyCount + UNDERSCORE
                             + getCompactForm(rpcDefinition);
                     synchronized (this)
                     {
                         this.keyCount++;
                     }
                     rpcCount.set(this.keyCount);
                     getRegistry().add(new StringField(keyName, rpcDefAsString));
                 }
                 synchronized (getHandlers())
                 {
                     getHandlers().put(keyName, handler);
                 }
                 synchronized (getPublished())
                 {
                     getPublished().put(keyName, rpcDefinition);
                 }
             }
             finally
             {
                 getRegistry().endFrame();
             }
             return true;
         }
         finally
         {
             // clear out the temporary key
             synchronized (getPublished())
             {
                 getPublished().remove(rpcDefAsString);
             }
         }
     }
 
     public boolean unpublishProdedure(IRpcDefinition rpcDefinition)
     {
         // remove the published RPC first
         synchronized (getPublished())
         {
             if (!getPublished().values().contains(rpcDefinition))
             {
                 return false;
             }
             final Set<Entry<String, IRpcDefinition>> entrySet =
                 getPublished().entrySet();
             for (Iterator<Entry<String, IRpcDefinition>> iterator =
                 entrySet.iterator(); iterator.hasNext();)
             {
                 Entry<String, IRpcDefinition> entry = iterator.next();
                 if (entry.getValue().equals(rpcDefinition))
                 {
                     iterator.remove();
                     break;
                 }
             }
         }
 
         getRegistry().beginFrame(new EventFrameExecution());
         try
         {
             final String[] fields = getRegistry().getComponentIdentities();
             // find the field that the RPC definition was declared at
             StringField rpcToRemove = null;
             for (String fieldName : fields)
             {
                 final IField field = getRegistry().get(fieldName);
                 if (field instanceof StringField)
                 {
                     final StringField key = (StringField) field;
                     if (key.get().equals(rpcDefinition.toString()))
                     {
                         // we've found the definition so this is the key to
                         // remove
                         rpcToRemove = key;
                         synchronized (getHandlers())
                         {
                             getHandlers().remove(key.getIdentity());
                         }
                         break;
                     }
                 }
             }
             getRegistry().remove(rpcToRemove);
         }
         finally
         {
             getRegistry().endFrame();
         }
         return true;
     }
 
     public boolean unpublishRpcs(Class<?> definition, Object handler)
     {
         return new RpcPublisher().disableRemoteMethods(definition, handler,
             this);
     }
 
     public boolean publishRpcs(Class<?> definition, Object handler)
     {
         return new RpcPublisher().enableRemoteMethods(definition, handler, this);
     }
 
     public IRpcHandler getHandler(String registryKey)
     {
         synchronized (getHandlers())
         {
             return getHandlers().get(registryKey);
         }
     }
 
     public IRpcDefinition getDefinition(String registryKey)
     {
         synchronized (getPublished())
         {
             return getPublished().get(registryKey);
         }
     }
 
     public String getRegistryKey(String rpcName, IField[] args)
     {
         synchronized (getPublished())
         {
             // go through all published RPCs to find a matching signature
             final Iterator<Entry<String, IRpcDefinition>> iterator =
                 getPublished().entrySet().iterator();
             while (iterator.hasNext())
             {
                 final Entry<String, IRpcDefinition> next = iterator.next();
                 final IRpcDefinition definition = next.getValue();
                 if (definition.getName().equals(rpcName))
                 {
                     // name matches, now check the arguments
                     if (args != null
                         && definition.getArgumentTypes().length == args.length)
                     {
                         // assume matching until a mismatching arg is found
                         boolean found = true;
                         for (int i = 0; i < args.length; i++)
                         {
                             IField field = args[i];
                             if (!field.getClass().equals(
                                 definition.getArgumentTypes()[i]))
                             {
                                 // no match on argument index
                                 found = false;
                                 break;
                             }
                         }
                         if (found)
                         {
                             // return the RPC registry key
                             return next.getKey();
                         }
                     }
                 }
             }
             return null;
         }
     }
 
     @Override
     protected void doDestroy()
     {
         getRegistry().destroy();
     }
 
     @Override
     protected void doStart()
     {
         this.registry =
             getContext().getLocalContainer(IRpcRegistry.RPC_REGISTRY,
                 Type.SYSTEM, Domain.FRAMEWORK);
         getRegistry().add(new IntegerField(IRpcRegistry.RPC_KEY_COUNT, 0));
     }
 
     /**
      * Get the compact form of the RPC definition. This is the RPC name followed
      * by the first letter of each argument type.
      * 
      * @param rpcDefinition
      *            the definition
      * @return its compact form
      */
     String getCompactForm(IRpcDefinition rpcDefinition)
     {
         StringBuilder sb = new StringBuilder();
         sb.append(rpcDefinition.getName());
         final Class<? extends IField>[] argumentTypes =
             rpcDefinition.getArgumentTypes();
         for (Class<? extends IField> argType : argumentTypes)
         {
             sb.append(argType.getSimpleName().substring(0, 1));
         }
         return sb.toString();
     }
 
     IContainer getRegistry()
     {
         return this.registry;
     }
 
     private IFrameworkContext getContext()
     {
         return this.context;
     }
 
     private Map<String, IRpcDefinition> getPublished()
     {
         return this.published;
     }
 
     private Map<String, IRpcHandler> getHandlers()
     {
         return this.handlers;
     }
 
     @Override
     public String toString()
     {
         return string(this, (getRegistry() == null ? EMPTY_STRING
             : this.getRegistry().toDetailedString()));
     }
 }
 
 /**
  * A stateless helper class that publishes and unpublishes RPCs represented by
  * Java classes.
  * 
  * @author Ramon Servadei
  */
 final class RpcPublisher
 {
     private final static Log LOG = new AsyncLog(RpcPublisher.class);
 
     /** Construct the {@link RpcPublisher} */
     RpcPublisher()
     {
         super();
     }
 
     /**
      * @see IRpcManager#unpublishRpcs(Class, Object)
      */
     boolean disableRemoteMethods(Class<?> definition, Object handler,
         IRpcRegistry registry)
     {
         return doOperation(definition, handler, registry, false);
     }
 
     /**
      * @see IRpcManager#publishRpcs(Class, Object)
      */
     boolean enableRemoteMethods(Class<?> definition, Object handler,
         IRpcRegistry registry)
     {
         return doOperation(definition, handler, registry, true);
     }
 
     /**
      * Convenience method that contains all the shared logic and uses a boolean
      * to flag whether to issue a publish or unpublish method call.
      * 
      * @param definition
      *            the definition
      * @param handler
      *            the handler that will receive invocations
      * @param state
      *            the state object to use
      * @param publish
      *            whether to publish or not
      * @return <code>true</code> if the operation succeeds
      */
     private boolean doOperation(Class<?> definition, Object handler,
         IRpcRegistry registry, boolean publish)
     {
         if (!definition.isInterface())
         {
             throw new IllegalArgumentException(
                 "The definition must be an interface, got " + definition);
         }
         final Method[] methods = definition.getMethods();
         final Method[] handlerMethods = new Method[methods.length];
         // check the handler has a method for each method in the interface
         for (int i = 0; i < methods.length; i++)
         {
             try
             {
                 Method method = methods[i];
                 handlerMethods[i] =
                     handler.getClass().getMethod(method.getName(),
                         method.getParameterTypes());
             }
             catch (SecurityException e)
             {
                 throw new IllegalStateException(e);
             }
             catch (NoSuchMethodException e)
             {
                 throw new IllegalArgumentException(safeToString(handler)
                     + " does not implement all methods of " + definition, e);
             }
         }
         boolean result = true;
         // now go through each method in the handler and process it
         for (Method handlerMethod : handlerMethods)
         {
             try
             {
                 IRpcDefinition rpcDefinition =
                     new RpcDefinition(
                         FieldUtils.getFrameworkClass(handlerMethod.getReturnType()),
                         handlerMethod.getName(),
                         RpcUtils.convertToFramework(handlerMethod.getParameterTypes()));
                 RpcHandler rpcHandler =
                     new RpcHandler(handler, rpcDefinition, handlerMethod);
                 if (publish)
                 {
                     result &=
                         registry.publishProdedure(rpcHandler, rpcDefinition);
                 }
                 else
                 {
                     result &= registry.unpublishProdedure(rpcDefinition);
                 }
             }
             catch (Exception e)
             {
                 Utils.logException(LOG, "Could not publish " + handlerMethod, e);
             }
         }
         return result;
     }
 }
 
 /**
  * A class that wraps the handler object supplied to the
  * {@link IRpcManager#publishRpcs(Class, Object)} and invokes only a single
  * remotely enabled method on the delegate handler.
  * <p>
  * An {@link RpcHandler} is compared for equality using its internal members.
  * 
  * @author Ramon Servadei
  */
 final class RpcHandler implements IRpcHandler
 {
     private final static AsyncLog LOG = new AsyncLog(RpcHandler.class);
 
     /** The handler that contains the remotely enable method */
     final Object delegate;
 
     /** The RPC definition this handler supports */
     final IRpcDefinition rpcDefinition;
 
     /** The method to invoke on the delegate */
     private final Method method;
 
     /**
      * Construct the handler with the delegate object that will handle a single,
      * specific RPC.
      * 
      * @param delegate
      *            the actual object that will handle an RPC
      * @param rpcDefinition
      *            the definition for the RPs this handler will support
      * @method the method to invoke on the delegate. This method will
      *         semantically match the rpcDefinition argument.
      */
     RpcHandler(Object delegate, IRpcDefinition rpcDefinition, Method method)
     {
         super();
         Utils.nullCheck(delegate, "delegate was null");
         Utils.nullCheck(rpcDefinition, "rpcDefinition was null");
         Utils.nullCheck(method, "method was null");
         this.delegate = delegate;
         this.rpcDefinition = rpcDefinition;
         this.method = method;
     }
 
     public IRpcResult handle(IRpcDefinition rpcDefinition, IField... arguments)
     {
         if (!getRpcDefinition().equals(rpcDefinition))
         {
             throw new IllegalArgumentException(safeToString(getDelegate())
                 + " cannot handle " + safeToString(rpcDefinition));
         }
         Object invokeResult = null;
         String exception = null;
         try
         {
             // ensure we can access even private methods
             getMethod().setAccessible(true);
             invokeResult =
                 getMethod().invoke(getDelegate(),
                     RpcUtils.convertToNative(arguments));
         }
         catch (Exception e)
         {
             if (LOG.isWarnEnabled())
             {
                 LOG.warn("Could not invoke method " + getMethod()
                     + " with arguments " + Arrays.deepToString(arguments), e);
             }
             exception = e.toString();
         }
         return new RpcResult(exception == null,
             FieldUtils.createFromNative(invokeResult), exception);
     }
 
     @Override
     public String toString()
     {
         return string(this, getDelegate() + COMMA + getMethod() + COMMA
             + getRpcDefinition());
     }
 
     @Override
     public int hashCode()
     {
         final int prime = 31;
         int result = 1;
         result =
             prime * result + ((delegate == null) ? 0 : delegate.hashCode());
         result =
             prime * result
                 + ((getMethod() == null) ? 0 : getMethod().hashCode());
         result =
             prime * result
                 + ((rpcDefinition == null) ? 0 : rpcDefinition.hashCode());
         return result;
     }
 
     @Override
     public boolean equals(Object obj)
     {
         if (is.same(this, obj))
         {
             return true;
         }
         if (is.differentClass(this, obj))
         {
             return false;
         }
         RpcHandler other = (RpcHandler) obj;
         return is.eq(getDelegate(), other.getDelegate())
             && is.eq(getMethod(), other.getMethod())
             && is.eq(getRpcDefinition(), other.getRpcDefinition());
     }
 
     private Method getMethod()
     {
         return this.method;
     }
 
     private Object getDelegate()
     {
         return this.delegate;
     }
 
     private IRpcDefinition getRpcDefinition()
     {
         return this.rpcDefinition;
     }
 
 }