org.codehaus.wadi.group.impl
Class AbstractDispatcher

java.lang.Object
  org.codehaus.wadi.group.impl.AbstractDispatcher

All Implemented Interfaces:

Dispatcher, EnvelopeListener

Direct Known Subclasses:

VMDispatcher

public abstract class AbstractDispatcher
extends java.lang.Object
implements Dispatcher

The portable aspects of a Dispatcher implementation

Version:

$Revision: 1595 $

Field Summary

protected ThreadPool	_executor
protected org.codehaus.wadi.group.impl.AbstractDispatcher.SimpleCorrelationIDFactory	_factory
protected org.apache.commons.logging.Log	_log
protected java.util.Map	_rvMap
protected java.util.List	interceptors

Constructor Summary

AbstractDispatcher()

AbstractDispatcher(ThreadPool executor)

Method Summary

void	addInterceptor(EnvelopeInterceptor interceptor)
void	addRendezVousEnvelope(Envelope envelope)
java.util.Collection	attemptMultiRendezVous(Quipu rv, long timeout)
Envelope	attemptRendezVous(Quipu rv, long timeout)
protected void	doOnEnvelope(Envelope envelope)
protected abstract void	doSend(Address target, Envelope envelope)
Envelope	exchangeSend(Address target, Envelope envelope, long timeout)
Envelope	exchangeSend(Address target, Envelope envelope, long timeout, java.lang.String targetCorrelationId)
Envelope	exchangeSend(Address to, java.io.Serializable body, long timeout) Send a Serializable 'pojo' to 'target' Address - sync - and wait for a reply.
Envelope	exchangeSend(Address target, java.io.Serializable pojo, long timeout, java.lang.String targetCorrelationId) Send a Serializable 'pojo' to a 'target' Address - sync - and wait for a reply.
DispatcherContext	getContext()
ThreadPool	getExecutor()
protected void	hook()
Quipu	newRendezVous(int numLlamas)
void	onEnvelope(Envelope envelope)
protected Envelope	onInboundEnvelope(Envelope envelope)
protected Envelope	onOutboundEnvelope(Envelope envelope)
void	register(ServiceEndpoint msgDispatcher)
void	removeInterceptor(EnvelopeInterceptor interceptor)
void	reply(Address from, Address to, java.lang.String incomingCorrelationId, java.io.Serializable body)
void	reply(Envelope request, Envelope reply)
void	reply(Envelope envelope, java.io.Serializable body)
void	send(Address source, Address target, Quipu quipu, java.io.Serializable pojo) send a Serializable 'pojo' to a 'target' Address, with the Message's replyTo field containing the 'source' Address - async - although we are expecting a reply, which will be matched with the sourceCorrelationId.
void	send(Address target, Envelope envelope) Send a ready-made Message to the Peer at the 'target' Address.
void	send(Address target, Quipu quipu, java.io.Serializable pojo) send a Serializable 'pojo' to a 'target' Address, with the Message's replyTo field containing the local cluster Address - async - although we are expecting a reply, which will be matched with the sourceCorrelationId.
void	send(Address to, java.io.Serializable body) send a Serializable pojo to an Address - async - no reply expected
protected Quipu	setRendezVous(java.lang.String correlationId, int numLlamas)
void	unregister(ServiceEndpoint msgDispatcher, int nbAttemp, long delayMillis)

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Methods inherited from interface org.codehaus.wadi.group.Dispatcher

createEnvelope, getCluster, getPeerName, start, stop

Field Detail

_executor

protected final ThreadPool _executor

_log

protected org.apache.commons.logging.Log _log

_rvMap

protected final java.util.Map _rvMap

interceptors

protected final java.util.List interceptors

_factory

protected final org.codehaus.wadi.group.impl.AbstractDispatcher.SimpleCorrelationIDFactory _factory

Constructor Detail

AbstractDispatcher

public AbstractDispatcher(ThreadPool executor)

AbstractDispatcher

public AbstractDispatcher()

Method Detail

getContext

public DispatcherContext getContext()

Specified by:

getContext in interface Dispatcher

addInterceptor

public void addInterceptor(EnvelopeInterceptor interceptor)

Specified by:

addInterceptor in interface Dispatcher

removeInterceptor

public void removeInterceptor(EnvelopeInterceptor interceptor)

Specified by:

removeInterceptor in interface Dispatcher

register

public void register(ServiceEndpoint msgDispatcher)

Specified by:

register in interface Dispatcher

unregister

public void unregister(ServiceEndpoint msgDispatcher,
                       int nbAttemp,
                       long delayMillis)

Specified by:

unregister in interface Dispatcher

onEnvelope

public final void onEnvelope(Envelope envelope)

Specified by:

onEnvelope in interface EnvelopeListener

doOnEnvelope

protected void doOnEnvelope(Envelope envelope)

addRendezVousEnvelope

public void addRendezVousEnvelope(Envelope envelope)

Specified by:

addRendezVousEnvelope in interface Dispatcher

newRendezVous

public Quipu newRendezVous(int numLlamas)

Specified by:

newRendezVous in interface Dispatcher

attemptRendezVous

public Envelope attemptRendezVous(Quipu rv,
                                  long timeout)
                           throws MessageExchangeException

Specified by:

attemptRendezVous in interface Dispatcher

Throws:

MessageExchangeException

attemptMultiRendezVous

public java.util.Collection attemptMultiRendezVous(Quipu rv,
                                                   long timeout)
                                            throws MessageExchangeException

Specified by:

attemptMultiRendezVous in interface Dispatcher

Throws:

MessageExchangeException

exchangeSend

public Envelope exchangeSend(Address to,
                             java.io.Serializable body,
                             long timeout)
                      throws MessageExchangeException

Description copied from interface: Dispatcher

Send a Serializable 'pojo' to 'target' Address - sync - and wait for a reply. The outgoing message will be accompanied by a correlation id. An incoming message, carrying the same correlation id, arriving within the specified timeframe, will be taken as the response.

Specified by:

exchangeSend in interface Dispatcher

Parameters:

to - The address to which we are sending the Message

body - The object that we want to send inside it

timeout - The length of time that we are willing to wait for a reply

Returns:

a response in the form of a Message

Throws:

MessageExchangeException

reply

public void reply(Envelope envelope,
                  java.io.Serializable body)
           throws MessageExchangeException

Specified by:

reply in interface Dispatcher

Throws:

MessageExchangeException

reply

public void reply(Envelope request,
                  Envelope reply)
           throws MessageExchangeException

Specified by:

reply in interface Dispatcher

Throws:

MessageExchangeException

send

public void send(Address to,
                 java.io.Serializable body)
          throws MessageExchangeException

Description copied from interface: Dispatcher

send a Serializable pojo to an Address - async - no reply expected

Specified by:

send in interface Dispatcher

Parameters:

to - The Address to which we are sending the Message

body - The Object that we want to send inside it

Throws:

MessageExchangeException

send

public void send(Address target,
                 Quipu quipu,
                 java.io.Serializable pojo)
          throws MessageExchangeException

Description copied from interface: Dispatcher

send a Serializable 'pojo' to a 'target' Address, with the Message's replyTo field containing the local cluster Address - async - although we are expecting a reply, which will be matched with the sourceCorrelationId. The code that calls this method assumes responsibility for coordinating this thread with the expected reply.

Specified by:

send in interface Dispatcher

Parameters:

target - The Address of the Peer to which this message is to be sent

quipu - The quipu which will be used to match a response on the source Peer

pojo - The object to be sent in the Message

Throws:

MessageExchangeException

send

public void send(Address source,
                 Address target,
                 Quipu quipu,
                 java.io.Serializable pojo)
          throws MessageExchangeException

Description copied from interface: Dispatcher

send a Serializable 'pojo' to a 'target' Address, with the Message's replyTo field containing the 'source' Address - async - although we are expecting a reply, which will be matched with the sourceCorrelationId. The code that calls this method assumes responsibility for coordinating this thread with the expected reply.

Specified by:

send in interface Dispatcher

Parameters:

source - The Address of the Peer to which the reply should be sent

target - The Address of the Peer to which this message is to be sent

quipu - The correlationId which will be used to match a response on the source Peer

pojo - The object to be sent in the Message

Throws:

MessageExchangeException

send

public final void send(Address target,
                       Envelope envelope)
                throws MessageExchangeException

Description copied from interface: Dispatcher

Send a ready-made Message to the Peer at the 'target' Address.

Specified by:

send in interface Dispatcher

Parameters:

target - The Address of the Peer to which the Message should be sent

envelope - The Message itself

Throws:

MessageExchangeException

exchangeSend

public Envelope exchangeSend(Address target,
                             java.io.Serializable pojo,
                             long timeout,
                             java.lang.String targetCorrelationId)
                      throws MessageExchangeException

Description copied from interface: Dispatcher

Send a Serializable 'pojo' to a 'target' Address - sync - and wait for a reply. This message will be accompanied by a 'targetCorrelationId' which will be used to match it at the target end, so that it may itself be interpreted as an incoming response to a previously outgoing request, thus allowing us to create 'message chains'.

Specified by:

exchangeSend in interface Dispatcher

Parameters:

target - The address to which we are sending the message

pojo - The object that we wish to send

timeout - The length of time that we are willing to wait for a reply

targetCorrelationId - Explicitly identifies the message to which we wish to reply

Returns:

a response in the form of a Message

Throws:

MessageExchangeException

exchangeSend

public Envelope exchangeSend(Address target,
                             Envelope envelope,
                             long timeout)
                      throws MessageExchangeException

Specified by:

exchangeSend in interface Dispatcher

Throws:

MessageExchangeException

exchangeSend

public Envelope exchangeSend(Address target,
                             Envelope envelope,
                             long timeout,
                             java.lang.String targetCorrelationId)
                      throws MessageExchangeException

Specified by:

exchangeSend in interface Dispatcher

Throws:

MessageExchangeException

reply

public void reply(Address from,
                  Address to,
                  java.lang.String incomingCorrelationId,
                  java.io.Serializable body)
           throws MessageExchangeException

Specified by:

reply in interface Dispatcher

Throws:

MessageExchangeException

setRendezVous

protected Quipu setRendezVous(java.lang.String correlationId,
                              int numLlamas)

hook

protected void hook()

getExecutor

public ThreadPool getExecutor()

doSend

protected abstract void doSend(Address target,
                               Envelope envelope)
                        throws MessageExchangeException

Throws:

MessageExchangeException

onOutboundEnvelope

protected Envelope onOutboundEnvelope(Envelope envelope)

onInboundEnvelope

protected Envelope onInboundEnvelope(Envelope envelope)