com.sun.sgs.impl.service.channel
Class ChannelServiceImpl

java.lang.Object
  com.sun.sgs.impl.util.AbstractService
      com.sun.sgs.impl.service.channel.ChannelServiceImpl

All Implemented Interfaces:

ChannelManager, Service

public final class ChannelServiceImpl
extends AbstractService
implements ChannelManager

ChannelService implementation.

The constructor requires the com.sun.sgs.app.name property and also supports the following additional properties:

Property: "com.sun.sgs.impl.service.channel.events.per.txn"
Default: 1

Specifies the maximum number of events to process in a single transaction.

Property: "com.sun.sgs.impl.service.channel.server.port"
Default: 0

Specifies the port to use for the ChannelServer of the local node.

Property: "com.sun.sgs.impl.service.channel.write.buffer.size"
Default: 131072

Specifies the approximate write buffer capacity per channel.

Property: "com.sun.sgs.impl.service.session.relocation.timeout"
Default: 10000L

Specifies the timeout, in milliseconds, for client session relocation. This also specifies the amount of time to save reliable channel messages so that relocating client sessions can obtain channel messages that were missed during relocation.

Nested Class Summary

(package private) class	ChannelServiceImpl.Context This transaction context maintains a per-channel list of non-transactional tasks to perform when the transaction commits.
(package private) static class	ChannelServiceImpl.MembershipEventType Channel membership event types.

Nested classes/interfaces inherited from class com.sun.sgs.impl.util.AbstractService

AbstractService.State, AbstractService.Version

Field Summary

(package private) static int	DEFAULT_EVENTS_PER_TXN The default events per transaction: 1.
(package private) static int	DEFAULT_SERVER_PORT The default server port: 0.
(package private) static int	DEFAULT_WRITE_BUFFER_SIZE The default write buffer size: 131072.
(package private) static String	EVENTS_PER_TXN_PROPERTY The property name for the maximum number of events to process in a single transaction.
(package private) int	eventsPerTxn The maximum number of channel events to service per transaction.
(package private) static String	SERVER_PORT_PROPERTY The name of the server port property.
(package private) ChannelServiceStats	serviceStats Our JMX exposed statistics.
(package private) long	sessionRelocationTimeout The timeout expiration, in milliseconds, for a client session to relocate.
(package private) static String	WRITE_BUFFER_SIZE_PROPERTY The name of the write buffer size property.

Fields inherited from class com.sun.sgs.impl.util.AbstractService

appName, dataService, IO_TASK_PROPERTY_PREFIX, IO_TASK_RETRIES_PROPERTY, IO_TASK_WAIT_TIME_PROPERTY, maxIoAttempts, retryWaitTime, taskOwner, taskScheduler, transactionScheduler, txnProxy

Constructor Summary

ChannelServiceImpl(Properties properties, ComponentRegistry systemRegistry, TransactionProxy txnProxy)
Constructs an instance of this class with the specified properties, systemRegistry, and txnProxy.

Method Summary

(package private) void	addChannelTaskOnCommit(BigInteger channelRefId, IoRunnable ioTask, long nodeId) Adds the specified ioTask (in a wrapper that runs the task by invoking runIoTask with the ioTask and nodeId) to the task list of the given channelRefId (when the current transaction commits).
(package private) void	addChannelTaskOnCommit(BigInteger channelRefId, KernelRunnable task) Adds the specified non-transactional task to the task list of the given channelRefId (when the current transaction commits).
(package private) void	addServiceEventQueueTask(BigInteger channelRefId) Adds a task to service the event queue of the channel with the specified channelRefId.
(package private) void	addServiceEventQueueTaskOnCommit(BigInteger channelRefId) Adds the specified channelRefId to the list of locally-coordinated channels that need servicing after the current transaction commits.
(package private) void	cacheMembershipEvent(ChannelServiceImpl.MembershipEventType eventType, BigInteger channelRefId, BigInteger sessionRefId, long eventTimestamp, long expirationTimestamp) Caches the channel membership event with the specified eventType, channelRefId, sessionRefId, and eventTimestamp.
(package private) static void	checkTransaction(Transaction txn) Checks that the specified context is currently active, throwing TransactionNotActiveException if it isn't.
(package private) void	closedChannel(BigInteger channelRefId) Notifies this service that the channel with the specified channelRefId is closed so that this service can clean up any per-channel data structures (relating to the channel coordinator).
(package private) Set<BigInteger>	collectChannelMembership(Transaction txn, BigInteger channelRefId, Set<Long> nodeIds) Collects a snapshot of the channel membership for the channel with the specified channelRefId and set of member nodeIds and returns an unmodifiable set containing the channel membership.
Channel	createChannel(String name, ChannelListener listener, Delivery delivery) Creates a new channel with the specified listener and delivery guarantee, binds it to the specified name, and returns it.
protected void	doReady() Performs ready operations.
protected void	doShutdown() Performs shutdown operations.
Channel	getChannel(String name) Returns an existing channel with the specified name.
(package private) ChannelServer	getChannelServer(long nodeId) Returns the ChannelServer for the given nodeId, or null if no channel server exists for the given nodeId.
(package private) static ClientSessionService	getClientSessionService() Returns the client session service.
(package private) static BindingKeyedCollections	getCollectionsFactory() Returns the BindingKeyedCollections instance.
(package private) static ChannelServiceImpl	getInstance() Returns the channel service.
(package private) static long	getLocalNodeId() Returns the local node ID.
(package private) static Object	getObjectForId(BigInteger refId) Returns the managed object with the specified refId, or null if there is no object with the specified refId.
(package private) static TaskService	getTaskService() Returns the task service.
(package private) static Transaction	getTransaction() Returns the currently active transaction, or throws TransactionNotActiveException if no transaction is active.
(package private) static WatchdogService	getWatchdogService() Returns the watchdog service.
void	handleChannelMessage(BigInteger channelRefId, ClientSession session, ByteBuffer message) Handles a channel message that the specified session is sending on the channel with the specified channelRefId.
protected void	handleServiceVersionMismatch(AbstractService.Version oldVersion, AbstractService.Version currentVersion) Handles conversion from the oldVersion to the currentVersion.
(package private) boolean	isChannelMember(BigInteger channelRefId, BigInteger sessionRefId, boolean isChannelMember, long timestamp) Returns true if the session with the specified sessionRefId is a member of the channel with the specified channelRefId, and false otherwise.
(package private) boolean	isLocalChannelMember(BigInteger channelRefId, BigInteger sessionRefId) Returns true if the session with the specified sessionRefId is a local member of the channel with the specified channelRefId.
(package private) <R> R	runTransactionalCallable(KernelCallable<R> callable) Runs the specified non-durable, transactional callable using this service's task owner, and returns the result.
(package private) void	runTransactionalTask(KernelRunnable task) Runs the specified non-durable, transactional task using this service's task owner.

Methods inherited from class com.sun.sgs.impl.util.AbstractService

callFinished, callStarted, checkNonTransactionalContext, checkServiceVersion, createTaskQueue, getDataService, getName, getState, isAlive, isInInitializedState, isRetryableException, ready, runIoTask, shutdown, shuttingDown

Methods inherited from class java.lang.Object

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

Field Detail

SERVER_PORT_PROPERTY

static final String SERVER_PORT_PROPERTY

The name of the server port property.

See Also:

Constant Field Values

DEFAULT_SERVER_PORT

static final int DEFAULT_SERVER_PORT

The default server port: 0.

See Also:

Constant Field Values

EVENTS_PER_TXN_PROPERTY

static final String EVENTS_PER_TXN_PROPERTY

The property name for the maximum number of events to process in a single transaction.

See Also:

Constant Field Values

DEFAULT_EVENTS_PER_TXN

static final int DEFAULT_EVENTS_PER_TXN

The default events per transaction: 1.

See Also:

Constant Field Values

WRITE_BUFFER_SIZE_PROPERTY

static final String WRITE_BUFFER_SIZE_PROPERTY

The name of the write buffer size property.

See Also:

Constant Field Values

DEFAULT_WRITE_BUFFER_SIZE

static final int DEFAULT_WRITE_BUFFER_SIZE

The default write buffer size: 131072.

See Also:

Constant Field Values

eventsPerTxn

final int eventsPerTxn

The maximum number of channel events to service per transaction.

sessionRelocationTimeout

final long sessionRelocationTimeout

The timeout expiration, in milliseconds, for a client session to relocate.

serviceStats

final ChannelServiceStats serviceStats

Our JMX exposed statistics.

Constructor Detail

ChannelServiceImpl

public ChannelServiceImpl(Properties properties,
                          ComponentRegistry systemRegistry,
                          TransactionProxy txnProxy)
                   throws Exception

Constructs an instance of this class with the specified properties, systemRegistry, and txnProxy.

Parameters:

properties - service properties

systemRegistry - system registry

txnProxy - transaction proxy

Throws:

Exception - if a problem occurs when creating the service

Method Detail

handleServiceVersionMismatch

protected void handleServiceVersionMismatch(AbstractService.Version oldVersion,
                                            AbstractService.Version currentVersion)

Handles conversion from the oldVersion to the currentVersion. This method is invoked by checkServiceVersion if a version mismatch is detected and is invoked from within a transaction.

Specified by:

handleServiceVersionMismatch in class AbstractService

Parameters:

oldVersion - the old version

currentVersion - the current version

doReady

protected void doReady()

Performs ready operations. This method is invoked by the ready method only once so that the subclass can perform any operations necessary during the "ready" phase.

Specified by:

doReady in class AbstractService

doShutdown

protected void doShutdown()

Performs shutdown operations. This method is invoked by the shutdown method only once so that the subclass can perform any operations necessary to shutdown the service.

Specified by:

doShutdown in class AbstractService

createChannel

public Channel createChannel(String name,
                             ChannelListener listener,
                             Delivery delivery)

Creates a new channel with the specified listener and delivery guarantee, binds it to the specified name, and returns it.

If the specified listener is null, then any messages sent on the returned channel by any client session will be automatically forwarded, unfiltered, to all channel members.

If the specified listener is non-null, then when any client session sends a message on the returned channel, the specified listener's receivedMessage method is invoked with the channel, client session and the message. The specified listener is not invoked for messages that the server sends on the channel via the channel's send method. If the specified listener is non-null, then it must also be Serializable.

Supplying a non-null listener (although not required) is strongly suggested. A listener's receivedMessage method provides an opportunity for an application to intervene when a client sends a channel message, to perform access control, filtering, or take other application-specific action on such channel messages.

If a non-null listener is provided, it is strongly suggested that a different listener instance be provided for each channel created, in order to reduce the possible contention on channel listeners.

Messages sent on the returned channel are delivered in a manner that satisfies the minimum requirements of the specified delivery guarantee.

Specified by:

createChannel in interface ChannelManager

Parameters:

name - a name

listener - a channel listener, or null

delivery - a delivery guarantee

Returns:

a new channel bound to the specified name

getChannel

public Channel getChannel(String name)

Returns an existing channel with the specified name.

Specified by:

getChannel in interface ChannelManager

Parameters:

name - a channel name

Returns:

an existing channel bound to the specified name

handleChannelMessage

public void handleChannelMessage(BigInteger channelRefId,
                                 ClientSession session,
                                 ByteBuffer message)

Handles a channel message that the specified session is sending on the channel with the specified channelRefId. This method is invoked from the ClientSessionHandler of the given session, when it receives a channel message. This method must be called from within a transaction.

Parameters:

channelRefId - the channel ID, as a BigInteger

session - the client session sending the channel message

message - the channel message

isLocalChannelMember

boolean isLocalChannelMember(BigInteger channelRefId,
                             BigInteger sessionRefId)

Returns true if the session with the specified sessionRefId is a local member of the channel with the specified channelRefId.

Parameters:

channelRefId - a channel ID

sessionRefId - a session ID

Returns:

true if the session with the specified sessionRefId is a local member of the channel with the specified channelRefId

collectChannelMembership

Set<BigInteger> collectChannelMembership(Transaction txn,
                                         BigInteger channelRefId,
                                         Set<Long> nodeIds)

Collects a snapshot of the channel membership for the channel with the specified channelRefId and set of member nodeIds and returns an unmodifiable set containing the channel membership.

Returns:

an unmodifiable set containing the channel membership

getTransaction

static Transaction getTransaction()

Returns the currently active transaction, or throws TransactionNotActiveException if no transaction is active.

checkTransaction

static void checkTransaction(Transaction txn)

Checks that the specified context is currently active, throwing TransactionNotActiveException if it isn't.

addChannelTaskOnCommit

void addChannelTaskOnCommit(BigInteger channelRefId,
                            IoRunnable ioTask,
                            long nodeId)

Adds the specified ioTask (in a wrapper that runs the task by invoking runIoTask with the ioTask and nodeId) to the task list of the given channelRefId (when the current transaction commits).

addChannelTaskOnCommit

void addChannelTaskOnCommit(BigInteger channelRefId,
                            KernelRunnable task)

Adds the specified non-transactional task to the task list of the given channelRefId (when the current transaction commits).

Parameters:

channelRefId - a channel ID

task - a non-transactional task

addServiceEventQueueTaskOnCommit

void addServiceEventQueueTaskOnCommit(BigInteger channelRefId)

Adds the specified channelRefId to the list of locally-coordinated channels that need servicing after the current transaction commits.

Parameters:

channelRefId - a channel ID for a locally-coordinated channel

closedChannel

void closedChannel(BigInteger channelRefId)

Notifies this service that the channel with the specified channelRefId is closed so that this service can clean up any per-channel data structures (relating to the channel coordinator).

getInstance

static ChannelServiceImpl getInstance()

Returns the channel service.

getClientSessionService

static ClientSessionService getClientSessionService()

Returns the client session service.

getTaskService

static TaskService getTaskService()

Returns the task service.

getWatchdogService

static WatchdogService getWatchdogService()

Returns the watchdog service.

getCollectionsFactory

static BindingKeyedCollections getCollectionsFactory()

Returns the BindingKeyedCollections instance.

getLocalNodeId

static long getLocalNodeId()

Returns the local node ID.

getChannelServer

ChannelServer getChannelServer(long nodeId)

Returns the ChannelServer for the given nodeId, or null if no channel server exists for the given nodeId. If the specified nodeId is the local node's ID, then this method returns a reference to the server implementation object, rather than the proxy.

runTransactionalTask

void runTransactionalTask(KernelRunnable task)
                    throws Exception

Runs the specified non-durable, transactional task using this service's task owner.

Parameters:

task - a transactional task

Throws:

Exception - the exception thrown while running task

runTransactionalCallable

<R> R runTransactionalCallable(KernelCallable<R> callable)
                     throws Exception

Runs the specified non-durable, transactional callable using this service's task owner, and returns the result.

Parameters:

callable - a callable to call

Throws:

Exception - the exception thrown while calliing callable

getObjectForId

static Object getObjectForId(BigInteger refId)

Returns the managed object with the specified refId, or null if there is no object with the specified refId.

Parameters:

refId - the object's identifier as obtained by ManagedReference.getId

Throws:

TransactionException - if the operation failed because of a problem with the current transaction

addServiceEventQueueTask

void addServiceEventQueueTask(BigInteger channelRefId)

Adds a task to service the event queue of the channel with the specified channelRefId. This method is only invoked on the channel's coordinator node.

Parameters:

channelRefId - a channel ID

cacheMembershipEvent

void cacheMembershipEvent(ChannelServiceImpl.MembershipEventType eventType,
                          BigInteger channelRefId,
                          BigInteger sessionRefId,
                          long eventTimestamp,
                          long expirationTimestamp)

Caches the channel membership event with the specified eventType, channelRefId, sessionRefId, and eventTimestamp. This method is only invoked on the channel's coordinator node.

Parameters:

eventType - an membership event type

channelRefId - a channel ID

sessionRefId - a session ID, or null

eventTimestamp - the event's timestamp

expirationTimestamp - the event queue's timestamp

isChannelMember

boolean isChannelMember(BigInteger channelRefId,
                        BigInteger sessionRefId,
                        boolean isChannelMember,
                        long timestamp)

Returns true if the session with the specified sessionRefId is a member of the channel with the specified channelRefId, and false otherwise. This method is only invoked on the channel's coordinator node.

Parameters:

channelRefId - a channel ID

sessionRefId - a session ID

isChannelMember - if true, the specified session is considered to be a member when current event was added to the event queue

timestamp - the timestamp of the currently executing event, beyond which join/leave requests should not be considered in determining channel membership

Returns:

true if the session with the specified sessionRefId is a member of the channel with the specified channelRefId, and false otherwise