com.bigdata.ha.pipeline

Class HAReceiveService<M extends HAMessageWrapper>

java.lang.Object

java.lang.Thread

com.bigdata.ha.pipeline.HAReceiveService<M>

All Implemented Interfaces:

Runnable

public class HAReceiveService<M extends HAMessageWrapper>
extends Thread

Receives data from an HASendService.

The non-blocking processing of the data cannot proceed until the message parameters and an output buffer have been set. So an accept results in a task to be run. The Future from this task is returned to the method called from the RMI control invocation, thus allowing that method to wait for the completion of the data transfer.

Author:

Martyn Cutcher, Bryan Thompson

Nested Class Summary

Nested Classes

Modifier and Type	Class and Description
static interface	HAReceiveService.IHAReceiveCallback<M extends IHAWriteMessageBase> Hook to notice receive events.

Nested classes/interfaces inherited from class java.lang.Thread

Thread.State, Thread.UncaughtExceptionHandler

Field Summary

Fields inherited from class java.lang.Thread

MAX_PRIORITY, MIN_PRIORITY, NORM_PRIORITY

Constructor Summary

Constructors

Constructor and Description
HAReceiveService(InetSocketAddress addrSelf, InetSocketAddress addrNext) Create a new service instance - you MUST Thread.start() the service.
HAReceiveService(InetSocketAddress addrSelf, InetSocketAddress addrNext, HAReceiveService.IHAReceiveCallback<M> callback) Create a new service instance - you MUST Thread.start() the service.

Method Summary

Methods

Modifier and Type	Method and Description
void	awaitShutdown() Block until the service is shutdown.
void	changeDownStream(InetSocketAddress addrNext) Change the address to which the payloads are being relayed.
void	changeUpStream() Method must be invoked when the upstream service is changed.
protected void	finalize() Extended to terminate() processing in order to ensure that the service is eventually shutdown.
InetSocketAddress	getAddrNext() The Internet socket address to which this service will relay messages (dynamic and MAY be null).
InetSocketAddress	getAddrSelf() The Internet socket address at which this service will listen (immutable)
HASendService	getSendService()
Future<Void>	receiveData(M msg, ByteBuffer buffer) Receive data into the caller's buffer as described by the caller's message.
void	run()
void	start()
void	terminate() Immediate shutdown.
String	toString()

Methods inherited from class java.lang.Thread

activeCount, checkAccess, clone, countStackFrames, currentThread, destroy, dumpStack, enumerate, getAllStackTraces, getContextClassLoader, getDefaultUncaughtExceptionHandler, getId, getName, getPriority, getStackTrace, getState, getThreadGroup, getUncaughtExceptionHandler, holdsLock, interrupt, interrupted, isAlive, isDaemon, isInterrupted, join, join, join, resume, setContextClassLoader, setDaemon, setDefaultUncaughtExceptionHandler, setName, setPriority, setUncaughtExceptionHandler, sleep, sleep, stop, stop, suspend, yield

Methods inherited from class java.lang.Object

equals, getClass, hashCode, notify, notifyAll, wait, wait, wait

Constructor Detail

HAReceiveService

public HAReceiveService(InetSocketAddress addrSelf,
                InetSocketAddress addrNext)

Create a new service instance - you MUST Thread.start() the service.

Note: addrNext can be changed dynamically.

Parameters:

addrSelf - The Internet socket address at which this service will listen.

addrNext - The Internet socket address of a downstream service to which each data transfer will be relayed as it is received (optional).

See Also:

changeDownStream(InetSocketAddress)

HAReceiveService

public HAReceiveService(InetSocketAddress addrSelf,
                InetSocketAddress addrNext,
                HAReceiveService.IHAReceiveCallback<M> callback)

Create a new service instance - you MUST Thread.start() the service.

Parameters:

addrSelf - The Internet socket address at which this service will listen.

addrNext - The Internet socket address of a downstream service to which each data transfer will be relayed as it is received (optional).

callback - An object which will be notified as each payload arrives.

Method Detail

getSendService

public HASendService getSendService()

toString

public String toString()

Overrides:

toString in class Thread

getAddrSelf

public InetSocketAddress getAddrSelf()

The Internet socket address at which this service will listen (immutable)

getAddrNext

public InetSocketAddress getAddrNext()

The Internet socket address to which this service will relay messages (dynamic and MAY be null).

See Also:

changeDownStream(InetSocketAddress)

finalize

protected void finalize()
                 throws Throwable

Extended to terminate() processing in order to ensure that the service is eventually shutdown.

Overrides:

finalize in class Object

Throws:

Throwable

terminate

public void terminate()

Immediate shutdown.

awaitShutdown

public void awaitShutdown()
                   throws InterruptedException

Block until the service is shutdown.

Throws:

InterruptedException

start

public void start()

Overrides:

start in class Thread

run

public void run()

Specified by:

run in interface Runnable

Overrides:

run in class Thread

receiveData

public Future<Void> receiveData(M msg,
                       ByteBuffer buffer)
                         throws InterruptedException

Receive data into the caller's buffer as described by the caller's message.

Parameters:

msg - The metadata about the data to be transferred.

buffer - The buffer in which this service will receive the data. The buffer MUST be large enough for the data to be received. The buffer SHOULD be a direct ByteBuffer in order to benefit from NIO efficiencies. This method will own the buffer until the returned Future is done.

Returns:

A Future which you can await. The Future will become available when the data has been transferred into the buffer, at which point the position will be ZERO (0) and the limit will be the #of bytes received into the buffer. If the data transfer fails or is interrupted, the Future will report the exception.

Throws:

InterruptedException

changeDownStream

public void changeDownStream(InetSocketAddress addrNext)

Change the address to which the payloads are being relayed. This terminates the embedded HASendService. The HASendService will be restarted with the new InetSocketAddress (if any) by the ReadTask.

Note: The ReadTask will throw out an exception when if there was a downstream target when the HASendService.IncSendTask is interrupted. Since the ReadTask lacks the context to issue the appropriate RMI to the downstream task, the exception must be caught and handled by the WriteCacheService. It can simply rediscover the new downstream service and then re-submit both the RMI and the WriteCache block.

Parameters:

addrNext - The new address -or- null if payloads should not be relayed at this time.

changeUpStream

public void changeUpStream()

Method must be invoked when the upstream service is changed. The method is responsible for interrupting the current RunTask (if any) and closing the client socket connection that was used to receive data from the upstream service. A new connection will be accepted by the next RunTask.