io.spine.server.delivery

Class AbstractWorkRegistry

java.lang.Object

io.spine.server.delivery.AbstractWorkRegistry

All Implemented Interfaces:

ShardedWorkRegistry

@SPI
public abstract class AbstractWorkRegistry
extends java.lang.Object
implements ShardedWorkRegistry

An implementation base for ShardedWorkRegistries based on a specific persistence mechanism.

Implementation Note:

This class is NOT thread safe. Synchronize the atomic persistence operations as well as the methods implemented in this class to make an implementation thread safe.

Constructor Summary

Constructors

Constructor and Description
AbstractWorkRegistry()

Method Summary

Modifier and Type	Method and Description
protected abstract java.util.Iterator<io.spine.server.delivery.ShardSessionRecord>	allRecords() Obtains all the session records associated with this registry.
protected abstract ShardProcessingSession	asSession(io.spine.server.delivery.ShardSessionRecord record) Restores a ShardProcessingSession from the given session record.
protected void	clearNode(io.spine.server.delivery.ShardSessionRecord session) Clears the value of ShardSessionRecord.worker and stores the session.
protected abstract io.spine.server.delivery.WorkerId	currentWorkerFor(io.spine.server.NodeId node) Returns an identifier of the current worker that is now going to process the shard.
protected abstract java.util.Optional<io.spine.server.delivery.ShardSessionRecord>	find(io.spine.server.delivery.ShardIndex index) Looks for the session record by the given shard index.
java.util.Optional<ShardProcessingSession>	pickUp(io.spine.server.delivery.ShardIndex index, io.spine.server.NodeId node) Picks up the shard at a given index to process.
java.lang.Iterable<io.spine.server.delivery.ShardIndex>	releaseExpiredSessions(com.google.protobuf.Duration inactivityPeriod) Clears up the recorded NodeIds from the session records if there was no activity for longer than passed inactivityPeriod.
protected abstract void	write(io.spine.server.delivery.ShardSessionRecord session) Stores the given session.

Methods inherited from class java.lang.Object

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

Constructor Detail

AbstractWorkRegistry

public AbstractWorkRegistry()

Method Detail

pickUp

public java.util.Optional<ShardProcessingSession> pickUp(io.spine.server.delivery.ShardIndex index,
                                                         io.spine.server.NodeId node)

Description copied from interface: ShardedWorkRegistry

Picks up the shard at a given index to process.

This action is intended to be exclusive, i.e. a single shard may be served by a single application node at a given moment of time.

In case of a successful operation, an instance of ShardProcessingSession is returned. The node obtained the session should perform the desired actions with the sharded messages and then complete the session.

In case the shard at a given index is already picked up by some node, an Optional.empty() is returned.

Specified by:

pickUp in interface ShardedWorkRegistry

Parameters:

index - the index of the shard to pick up for processing

node - the identifier of the node for which to pick the shard

Returns:

the session of shard processing, or Optional.empty() if the shard is not available

currentWorkerFor

protected abstract io.spine.server.delivery.WorkerId currentWorkerFor(io.spine.server.NodeId node)

Returns an identifier of the current worker that is now going to process the shard.

An example of such an identifier could be ID of the thread which performs processing.

Parameters:

node - the node to which the resulted worker belongs

releaseExpiredSessions

public java.lang.Iterable<io.spine.server.delivery.ShardIndex> releaseExpiredSessions(com.google.protobuf.Duration inactivityPeriod)

Description copied from interface: ShardedWorkRegistry

Clears up the recorded NodeIds from the session records if there was no activity for longer than passed inactivityPeriod.

It may be handy if an application node hangs or gets killed — so that it is not able to complete the session in a conventional way.

Specified by:

releaseExpiredSessions in interface ShardedWorkRegistry

Parameters:

inactivityPeriod - the duration of the period after which the session is considered expired

Returns:

the indexes of shards which sessions have been released

clearNode

protected void clearNode(io.spine.server.delivery.ShardSessionRecord session)

Clears the value of ShardSessionRecord.worker and stores the session.

allRecords

protected abstract java.util.Iterator<io.spine.server.delivery.ShardSessionRecord> allRecords()

Obtains all the session records associated with this registry.

write

protected abstract void write(io.spine.server.delivery.ShardSessionRecord session)

Stores the given session.

The session may or may be not present in the registry already. After calling this method, the given session must be reachable via AbstractWorkRegistry.find(ShardIndex) and AbstractWorkRegistry.allRecords().

find

protected abstract java.util.Optional<io.spine.server.delivery.ShardSessionRecord> find(io.spine.server.delivery.ShardIndex index)

Looks for the session record by the given shard index.

Parameters:

index - shard index to find a session for

Returns:

a session record or Optional.empty() if the record is not present in the registry

asSession

protected abstract ShardProcessingSession asSession(io.spine.server.delivery.ShardSessionRecord record)

Restores a ShardProcessingSession from the given session record.