bundles/org.apache.tomcat/src/org/apache/catalina/tribes/group/RpcChannel.java - rap/org.apache.tomcat - Git at Google

 /*
  * Licensed to the Apache Software Foundation (ASF) under one or more
  * contributor license agreements.  See the NOTICE file distributed with
  * this work for additional information regarding copyright ownership.
  * The ASF licenses this file to You 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.apache.catalina.tribes.group;

 import java.io.Serializable;
 import java.util.ArrayList;
 import java.util.Arrays;
 import java.util.HashMap;

 import org.apache.catalina.tribes.Channel;
 import org.apache.catalina.tribes.ChannelException;
 import org.apache.catalina.tribes.ChannelListener;
 import org.apache.catalina.tribes.ErrorHandler;
 import org.apache.catalina.tribes.Member;
 import org.apache.catalina.tribes.UniqueId;
 import org.apache.catalina.tribes.util.UUIDGenerator;
 import org.apache.juli.logging.Log;
 import org.apache.juli.logging.LogFactory;

 /**
  * A channel to handle RPC messaging
  * @author Filip Hanik
  */
 public class RpcChannel implements ChannelListener{
     private static final Log log = LogFactory.getLog(RpcChannel.class);

     public static final int FIRST_REPLY = 1;
     public static final int MAJORITY_REPLY = 2;
     public static final int ALL_REPLY = 3;
     public static final int NO_REPLY = 4;

     private Channel channel;
     private RpcCallback callback;
     private byte[] rpcId;
     private int replyMessageOptions = 0;

     private HashMap<RpcCollectorKey, RpcCollector> responseMap = new HashMap<RpcCollectorKey, RpcCollector>();

     /**
      * Create an RPC channel. You can have several RPC channels attached to a group
      * all separated out by the uniqueness
      * @param rpcId - the unique Id for this RPC group
      * @param channel Channel
      * @param callback RpcCallback
      */
     public RpcChannel(byte[] rpcId, Channel channel, RpcCallback callback) {
         this.channel = channel;
         this.callback = callback;
         this.rpcId = rpcId;
         channel.addChannelListener(this);
     }


     /**
      * Send a message and wait for the response.
      * @param destination Member[] - the destination for the message, and the members you request a reply from
      * @param message Serializable - the message you are sending out
      * @param rpcOptions int - FIRST_REPLY, MAJORITY_REPLY or ALL_REPLY
      * @param channelOptions channel sender options
      * @param timeout long - timeout in milliseconds, if no reply is received within this time null is returned
      * @return Response[] - an array of response objects.
      * @throws ChannelException
      */
     public Response[] send(Member[] destination,
                            Serializable message,
                            int rpcOptions,
                            int channelOptions,
                            long timeout) throws ChannelException {

         if ( destination==null || destination.length == 0 ) return new Response[0];

         //avoid dead lock
         int sendOptions =
             channelOptions & ~Channel.SEND_OPTIONS_SYNCHRONIZED_ACK;

         RpcCollectorKey key = new RpcCollectorKey(UUIDGenerator.randomUUID(false));
         RpcCollector collector = new RpcCollector(key,rpcOptions,destination.length,timeout);
         try {
             synchronized (collector) {
                 if ( rpcOptions != NO_REPLY ) responseMap.put(key, collector);
                 RpcMessage rmsg = new RpcMessage(rpcId, key.id, message);
                 channel.send(destination, rmsg, sendOptions);
                 if ( rpcOptions != NO_REPLY ) collector.wait(timeout);
             }
         } catch ( InterruptedException ix ) {
             Thread.currentThread().interrupt();
         }finally {
             responseMap.remove(key);
         }
         return collector.getResponses();
     }

     @Override
     public void messageReceived(Serializable msg, Member sender) {
         RpcMessage rmsg = (RpcMessage)msg;
         RpcCollectorKey key = new RpcCollectorKey(rmsg.uuid);
         if ( rmsg.reply ) {
             RpcCollector collector = responseMap.get(key);
             if (collector == null) {
                 callback.leftOver(rmsg.message, sender);
             } else {
                 synchronized (collector) {
                     //make sure it hasn't been removed
                     if ( responseMap.containsKey(key) ) {
                         if ( (rmsg instanceof RpcMessage.NoRpcChannelReply) )
                             collector.destcnt--;
                         else
                             collector.addResponse(rmsg.message, sender);
                         if (collector.isComplete()) collector.notifyAll();
                     } else {
                         if (! (rmsg instanceof RpcMessage.NoRpcChannelReply) )
                             callback.leftOver(rmsg.message, sender);
                     }
                 }//synchronized
             }//end if
         } else{
             boolean finished = false;
             final ExtendedRpcCallback excallback = (callback instanceof ExtendedRpcCallback)?((ExtendedRpcCallback)callback) : null;
             boolean asyncReply = ((replyMessageOptions & Channel.SEND_OPTIONS_ASYNCHRONOUS) == Channel.SEND_OPTIONS_ASYNCHRONOUS);
             Serializable reply = callback.replyRequest(rmsg.message,sender);
             ErrorHandler handler = null;
             final Serializable request = msg;
             final Serializable response = reply;
             final Member fsender = sender;
             if (excallback!=null && asyncReply) {
                 handler = new ErrorHandler() {
                     @Override
                     public void handleError(ChannelException x, UniqueId id) {
                         excallback.replyFailed(request, response, fsender, x);
                     }
                     @Override
                     public void handleCompletion(UniqueId id) {
                         excallback.replySucceeded(request, response, fsender);
                     }
                 };
             }
             rmsg.reply = true;
             rmsg.message = reply;
             try {
                 if (handler!=null) {
                     channel.send(new Member[] {sender}, rmsg,replyMessageOptions & ~Channel.SEND_OPTIONS_SYNCHRONIZED_ACK, handler);
                 } else {
                     channel.send(new Member[] {sender}, rmsg,replyMessageOptions & ~Channel.SEND_OPTIONS_SYNCHRONIZED_ACK);
                 }
                 finished = true;
             }catch ( Exception x )  {
                 if (excallback != null && !asyncReply) {
                     excallback.replyFailed(rmsg.message, reply, sender, x);
                 } else {
                     log.error("Unable to send back reply in RpcChannel.",x);
                 }
             }
             if (finished && excallback != null && !asyncReply) {
                 excallback.replySucceeded(rmsg.message, reply, sender);
             }
         }//end if
     }

     public void breakdown() {
         channel.removeChannelListener(this);
     }

     @Override
     public void finalize() {
         breakdown();
     }

     @Override
     public boolean accept(Serializable msg, Member sender) {
         if ( msg instanceof RpcMessage ) {
             RpcMessage rmsg = (RpcMessage)msg;
             return Arrays.equals(rmsg.rpcId,rpcId);
         }else return false;
     }

     public Channel getChannel() {
         return channel;
     }

     public RpcCallback getCallback() {
         return callback;
     }

     public byte[] getRpcId() {
         return rpcId;
     }

     public void setChannel(Channel channel) {
         this.channel = channel;
     }

     public void setCallback(RpcCallback callback) {
         this.callback = callback;
     }

     public void setRpcId(byte[] rpcId) {
         this.rpcId = rpcId;
     }

     public int getReplyMessageOptions() {
         return replyMessageOptions;
     }

     public void setReplyMessageOptions(int replyMessageOptions) {
         this.replyMessageOptions = replyMessageOptions;
     }

     /**
      *
      * Class that holds all response.
      * @author not attributable
      * @version 1.0
      */
     public static class RpcCollector {
         public ArrayList<Response> responses = new ArrayList<Response>();
         public RpcCollectorKey key;
         public int options;
         public int destcnt;
         public long timeout;

         public RpcCollector(RpcCollectorKey key, int options, int destcnt, long timeout) {
             this.key = key;
             this.options = options;
             this.destcnt = destcnt;
             this.timeout = timeout;
         }

         public void addResponse(Serializable message, Member sender){
             Response resp = new Response(sender,message);
             responses.add(resp);
         }

         public boolean isComplete() {
             if ( destcnt <= 0 ) return true;
             switch (options) {
                 case ALL_REPLY:
                     return destcnt == responses.size();
                 case MAJORITY_REPLY:
                 {
                     float perc = ((float)responses.size()) / ((float)destcnt);
                     return perc >= 0.50f;
                 }
                 case FIRST_REPLY:
                     return responses.size()>0;
                 default:
                     return false;
             }
         }

         @Override
         public int hashCode() {
             return key.hashCode();
         }

         @Override
         public boolean equals(Object o) {
             if ( o instanceof RpcCollector ) {
                 RpcCollector r = (RpcCollector)o;
                 return r.key.equals(this.key);
             } else return false;
         }

         public Response[] getResponses() {
             return responses.toArray(new Response[responses.size()]);
         }
     }

     public static class RpcCollectorKey {
         byte[] id;
         public RpcCollectorKey(byte[] id) {
             this.id = id;
         }

         @Override
         public int hashCode() {
             return id[0]+id[1]+id[2]+id[3];
         }

         @Override
         public boolean equals(Object o) {
             if ( o instanceof RpcCollectorKey ) {
                 RpcCollectorKey r = (RpcCollectorKey)o;
                 return Arrays.equals(id,r.id);
             } else return false;
         }

     }

     protected static String bToS(byte[] data) {
         StringBuilder buf = new StringBuilder(4*16);
         buf.append("{");
         for (int i=0; data!=null && i<data.length; i++ ) buf.append(String.valueOf(data[i])).append(" ");
         buf.append("}");
         return buf.toString();
     }


 }
	/*
	* Licensed to the Apache Software Foundation (ASF) under one or more
	* contributor license agreements. See the NOTICE file distributed with
	* this work for additional information regarding copyright ownership.
	* The ASF licenses this file to You 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.apache.catalina.tribes.group;

	import java.io.Serializable;
	import java.util.ArrayList;
	import java.util.Arrays;
	import java.util.HashMap;

	import org.apache.catalina.tribes.Channel;
	import org.apache.catalina.tribes.ChannelException;
	import org.apache.catalina.tribes.ChannelListener;
	import org.apache.catalina.tribes.ErrorHandler;
	import org.apache.catalina.tribes.Member;
	import org.apache.catalina.tribes.UniqueId;
	import org.apache.catalina.tribes.util.UUIDGenerator;
	import org.apache.juli.logging.Log;
	import org.apache.juli.logging.LogFactory;

	/**
	* A channel to handle RPC messaging
	* @author Filip Hanik
	*/
	public class RpcChannel implements ChannelListener{
	private static final Log log = LogFactory.getLog(RpcChannel.class);

	public static final int FIRST_REPLY = 1;
	public static final int MAJORITY_REPLY = 2;
	public static final int ALL_REPLY = 3;
	public static final int NO_REPLY = 4;

	private Channel channel;
	private RpcCallback callback;
	private byte[] rpcId;
	private int replyMessageOptions = 0;

	private HashMap<RpcCollectorKey, RpcCollector> responseMap = new HashMap<RpcCollectorKey, RpcCollector>();

	/**
	* Create an RPC channel. You can have several RPC channels attached to a group
	* all separated out by the uniqueness
	* @param rpcId - the unique Id for this RPC group
	* @param channel Channel
	* @param callback RpcCallback
	*/
	public RpcChannel(byte[] rpcId, Channel channel, RpcCallback callback) {
	this.channel = channel;
	this.callback = callback;
	this.rpcId = rpcId;
	channel.addChannelListener(this);
	}


	/**
	* Send a message and wait for the response.
	* @param destination Member[] - the destination for the message, and the members you request a reply from
	* @param message Serializable - the message you are sending out
	* @param rpcOptions int - FIRST_REPLY, MAJORITY_REPLY or ALL_REPLY
	* @param channelOptions channel sender options
	* @param timeout long - timeout in milliseconds, if no reply is received within this time null is returned
	* @return Response[] - an array of response objects.
	* @throws ChannelException
	*/
	public Response[] send(Member[] destination,
	Serializable message,
	int rpcOptions,
	int channelOptions,
	long timeout) throws ChannelException {

	if ( destination==null \|\| destination.length == 0 ) return new Response[0];

	//avoid dead lock
	int sendOptions =
	channelOptions & ~Channel.SEND_OPTIONS_SYNCHRONIZED_ACK;

	RpcCollectorKey key = new RpcCollectorKey(UUIDGenerator.randomUUID(false));
	RpcCollector collector = new RpcCollector(key,rpcOptions,destination.length,timeout);
	try {
	synchronized (collector) {
	if ( rpcOptions != NO_REPLY ) responseMap.put(key, collector);
	RpcMessage rmsg = new RpcMessage(rpcId, key.id, message);
	channel.send(destination, rmsg, sendOptions);
	if ( rpcOptions != NO_REPLY ) collector.wait(timeout);
	}
	} catch ( InterruptedException ix ) {
	Thread.currentThread().interrupt();
	}finally {
	responseMap.remove(key);
	}
	return collector.getResponses();
	}

	@Override
	public void messageReceived(Serializable msg, Member sender) {
	RpcMessage rmsg = (RpcMessage)msg;
	RpcCollectorKey key = new RpcCollectorKey(rmsg.uuid);
	if ( rmsg.reply ) {
	RpcCollector collector = responseMap.get(key);
	if (collector == null) {
	callback.leftOver(rmsg.message, sender);
	} else {
	synchronized (collector) {
	//make sure it hasn't been removed
	if ( responseMap.containsKey(key) ) {
	if ( (rmsg instanceof RpcMessage.NoRpcChannelReply) )
	collector.destcnt--;
	else
	collector.addResponse(rmsg.message, sender);
	if (collector.isComplete()) collector.notifyAll();
	} else {
	if (! (rmsg instanceof RpcMessage.NoRpcChannelReply) )
	callback.leftOver(rmsg.message, sender);
	}
	}//synchronized
	}//end if
	} else{
	boolean finished = false;
	final ExtendedRpcCallback excallback = (callback instanceof ExtendedRpcCallback)?((ExtendedRpcCallback)callback) : null;
	boolean asyncReply = ((replyMessageOptions & Channel.SEND_OPTIONS_ASYNCHRONOUS) == Channel.SEND_OPTIONS_ASYNCHRONOUS);
	Serializable reply = callback.replyRequest(rmsg.message,sender);
	ErrorHandler handler = null;
	final Serializable request = msg;
	final Serializable response = reply;
	final Member fsender = sender;
	if (excallback!=null && asyncReply) {
	handler = new ErrorHandler() {
	@Override
	public void handleError(ChannelException x, UniqueId id) {
	excallback.replyFailed(request, response, fsender, x);
	}
	@Override
	public void handleCompletion(UniqueId id) {
	excallback.replySucceeded(request, response, fsender);
	}
	};
	}
	rmsg.reply = true;
	rmsg.message = reply;
	try {
	if (handler!=null) {
	channel.send(new Member[] {sender}, rmsg,replyMessageOptions & ~Channel.SEND_OPTIONS_SYNCHRONIZED_ACK, handler);
	} else {
	channel.send(new Member[] {sender}, rmsg,replyMessageOptions & ~Channel.SEND_OPTIONS_SYNCHRONIZED_ACK);
	}
	finished = true;
	}catch ( Exception x ) {
	if (excallback != null && !asyncReply) {
	excallback.replyFailed(rmsg.message, reply, sender, x);
	} else {
	log.error("Unable to send back reply in RpcChannel.",x);
	}
	}
	if (finished && excallback != null && !asyncReply) {
	excallback.replySucceeded(rmsg.message, reply, sender);
	}
	}//end if
	}

	public void breakdown() {
	channel.removeChannelListener(this);
	}

	@Override
	public void finalize() {
	breakdown();
	}

	@Override
	public boolean accept(Serializable msg, Member sender) {
	if ( msg instanceof RpcMessage ) {
	RpcMessage rmsg = (RpcMessage)msg;
	return Arrays.equals(rmsg.rpcId,rpcId);
	}else return false;
	}

	public Channel getChannel() {
	return channel;
	}

	public RpcCallback getCallback() {
	return callback;
	}

	public byte[] getRpcId() {
	return rpcId;
	}

	public void setChannel(Channel channel) {
	this.channel = channel;
	}

	public void setCallback(RpcCallback callback) {
	this.callback = callback;
	}

	public void setRpcId(byte[] rpcId) {
	this.rpcId = rpcId;
	}

	public int getReplyMessageOptions() {
	return replyMessageOptions;
	}

	public void setReplyMessageOptions(int replyMessageOptions) {
	this.replyMessageOptions = replyMessageOptions;
	}

	/**
	*
	* Class that holds all response.
	* @author not attributable
	* @version 1.0
	*/
	public static class RpcCollector {
	public ArrayList<Response> responses = new ArrayList<Response>();
	public RpcCollectorKey key;
	public int options;
	public int destcnt;
	public long timeout;

	public RpcCollector(RpcCollectorKey key, int options, int destcnt, long timeout) {
	this.key = key;
	this.options = options;
	this.destcnt = destcnt;
	this.timeout = timeout;
	}

	public void addResponse(Serializable message, Member sender){
	Response resp = new Response(sender,message);
	responses.add(resp);
	}

	public boolean isComplete() {
	if ( destcnt <= 0 ) return true;
	switch (options) {
	case ALL_REPLY:
	return destcnt == responses.size();
	case MAJORITY_REPLY:
	{
	float perc = ((float)responses.size()) / ((float)destcnt);
	return perc >= 0.50f;
	}
	case FIRST_REPLY:
	return responses.size()>0;
	default:
	return false;
	}
	}

	@Override
	public int hashCode() {
	return key.hashCode();
	}

	@Override
	public boolean equals(Object o) {
	if ( o instanceof RpcCollector ) {
	RpcCollector r = (RpcCollector)o;
	return r.key.equals(this.key);
	} else return false;
	}

	public Response[] getResponses() {
	return responses.toArray(new Response[responses.size()]);
	}
	}

	public static class RpcCollectorKey {
	byte[] id;
	public RpcCollectorKey(byte[] id) {
	this.id = id;
	}

	@Override
	public int hashCode() {
	return id[0]+id[1]+id[2]+id[3];
	}

	@Override
	public boolean equals(Object o) {
	if ( o instanceof RpcCollectorKey ) {
	RpcCollectorKey r = (RpcCollectorKey)o;
	return Arrays.equals(id,r.id);
	} else return false;
	}

	}

	protected static String bToS(byte[] data) {
	StringBuilder buf = new StringBuilder(4*16);
	buf.append("{");
	for (int i=0; data!=null && i<data.length; i++ ) buf.append(String.valueOf(data[i])).append(" ");
	buf.append("}");
	return buf.toString();
	}


	}