Skip navigation links

com.oracle.cep.cluster.ha.adapter.inbound
Class BroadcastInboundAdapter

java.lang.Object
  |
  +--com.oracle.cep.cluster.ha.adapter.inbound.BroadcastInboundAdapter

public class BroadcastInboundAdapter

extends java.lang.Object

implements HAAdapter, StreamSource, StreamSink, GroupMembershipListener, InitializingBean, DisposableBean, ActivatableBean, RunnableBean, ApplicationIdentityAware, StageIdentityAware, GroupNameAware

The BroadcastInboundAdapter is intended for applications that use system time and also need to be highly available. For a detailed discussion of system time and how it differs from application time, see the Oracle CEP product documentation. The BroadcastInboundAdapter should be configured as a stage downstream from the application's actual inbound adapter and upstream from EPN components that rely on events having a well-defined time. The BroadcastInboundAdapter assigns a time (in nanoseconds) to events as they arrive at the adapter and forwards the time values assigned to events to other servers hosting the highly available application. This ensures that all servers running the application use a consistent time value (and generate the same results) and avoids the need for distributed clock synchronization. Since a time value is assigned to each event before the event reaches any downstream channels in the EPN, downstream channels should be configured to use application time so that they do not assign a new time value to events as they arrive at the channel. Input events must have a key that uniquely identifies each event in order to use this adapter. The BroadcastInboundAdapter may be configured to send heartbeat events, as well. The BroadcastInboundAdapter supports a number of statically configurable instance-properties:

 timeProperty - name of event time property to which a value is assigned
 keyProperties - name of event key property (s)
 keyClass - Java class name of compound key class
 eventType - event type name for the events processed by this adapter 
             (optional if using Java events)
 

The following dynamic properties are supported. Changes take effect immediately.

 batch-size Specifies a batch size for timing messages sent from the primary.
            A value of n means that n (key, time) pairs are sent as a batch.
  Default value is 1 (disabled).
 heartbeat How often heartbeat messages should be sent. Units are 'nanos', 
           'millis', or 'secs'.  Heartbeats are disabled by default.
 

The BroadcastInboundAdapter is used by specifying the 'ha-inbound' provider.

Field Summary

static int	BECOMING_SECONDARY
protected java.util.Queue	eventQueue
protected java.util.concurrent.locks.ReentrantLock	insertLock
protected Log	log
static int	PRIMARY
protected java.util.concurrent.locks.ReentrantLock	queueLock
static int	SECONDARY
protected java.lang.String	stageIdentity
protected java.util.concurrent.atomic.AtomicInteger	state

Constructor Summary

BroadcastInboundAdapter()

Method Summary
void	activateConfiguration(BroadcastInboundAdapterConfig config)
void	afterConfigurationActive()
void	afterPropertiesSet()
void	checkConfiguration(BroadcastInboundAdapterConfig config)
void	destroy()
java.lang.String	getApplicationIdentity()
java.util.Queue	getEventQueue()
java.lang.String	getEventType()
EventTypeRepository	getEventTypeRepository()
java.lang.String	getGroupName()
long	getLastReceivedTimestamp()
java.util.concurrent.ConcurrentLinkedQueue	getMessageQueue()
long	getReceivedMessageCount()
long	getSentMessageCount()
java.lang.String	getStageIdentity()
int	getState()
java.lang.String	getTimeProperty()
boolean	isPrimary()
void	onInsertEvent(java.lang.Object inboundEvent)
void	onMembershipChange(Server server, Configuration configuration)
void	onMessage(java.lang.Object message)
void	rollbackConfiguration(BroadcastInboundAdapterConfig config)
void	run()
void	sendHeartbeatMessage(long heartbeatTimestamp)
void	setApplicationIdentity(java.lang.String applicationIdentity)
void	setBroadcastDomain(BroadcastDomain broadcastDomain)
void	setBundleContext(BundleContext bundleContext)
void	setEventSender(StreamSender sender)
void	setEventType(java.lang.String eventType)
void	setEventTypeRepository(EventTypeRepository eventTypeRepository)
void	setGroupName(java.lang.String groupName)
void	setKeyClass(java.lang.Class keyClass)
void	setKeyProperties(java.lang.String keyProperties)
void	setQueue(java.util.Queue queue) Set the TrimmableQueue for this adapter
void	setStageIdentity(java.lang.String stageIdentity)
void	setTimeProperty(java.lang.String timeProperty)
void	suspend()

Methods inherited from class java.lang.Object

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

Field Detail

PRIMARY

public static int PRIMARY

SECONDARY

public static int SECONDARY

BECOMING_SECONDARY

public static int BECOMING_SECONDARY

eventQueue

protected java.util.Queue eventQueue

state

protected java.util.concurrent.atomic.AtomicInteger state

log

protected Log log

stageIdentity

protected java.lang.String stageIdentity

queueLock

protected final java.util.concurrent.locks.ReentrantLock queueLock

insertLock

protected final java.util.concurrent.locks.ReentrantLock insertLock

Constructor Detail

BroadcastInboundAdapter

public BroadcastInboundAdapter()

Method Detail

setEventSender

public void setEventSender(StreamSender sender)

Specified by:

setEventSender in interface StreamSource

setKeyProperties

public void setKeyProperties(java.lang.String keyProperties)

setKeyClass

public void setKeyClass(java.lang.Class keyClass)

setQueue

public void setQueue(java.util.Queue queue)

Set the TrimmableQueue for this adapter

Parameters:

queue - to use for trimming.

setBroadcastDomain

public void setBroadcastDomain(BroadcastDomain broadcastDomain)

afterPropertiesSet

public void afterPropertiesSet()

Specified by:

afterPropertiesSet in interface InitializingBean

checkConfiguration

public void checkConfiguration(BroadcastInboundAdapterConfig config)

activateConfiguration

public void activateConfiguration(BroadcastInboundAdapterConfig config)

rollbackConfiguration

public void rollbackConfiguration(BroadcastInboundAdapterConfig config)

afterConfigurationActive

public void afterConfigurationActive()
                              throws java.lang.Exception

Specified by:

afterConfigurationActive in interface ActivatableBean

run

public void run()

suspend

public void suspend()
             throws java.lang.Exception

onInsertEvent

public void onInsertEvent(java.lang.Object inboundEvent)
                   throws EventRejectedException

Specified by:

onInsertEvent in interface StreamSink

sendHeartbeatMessage

public void sendHeartbeatMessage(long heartbeatTimestamp)

onMembershipChange

public void onMembershipChange(Server server,
                               Configuration configuration)

Specified by:

onMembershipChange in interface GroupMembershipListener

onMessage

public void onMessage(java.lang.Object message)

isPrimary

public boolean isPrimary()

Specified by:

isPrimary in interface HAAdapter

setApplicationIdentity

public void setApplicationIdentity(java.lang.String applicationIdentity)

Specified by:

setApplicationIdentity in interface ApplicationIdentityAware

setStageIdentity

public void setStageIdentity(java.lang.String stageIdentity)

Specified by:

setStageIdentity in interface StageIdentityAware

getApplicationIdentity

public java.lang.String getApplicationIdentity()

getStageIdentity

public java.lang.String getStageIdentity()

setGroupName

public void setGroupName(java.lang.String groupName)

Specified by:

setGroupName in interface GroupNameAware

getGroupName

public java.lang.String getGroupName()

getTimeProperty

public java.lang.String getTimeProperty()

setTimeProperty

public void setTimeProperty(java.lang.String timeProperty)

getEventQueue

public java.util.Queue getEventQueue()

getMessageQueue

public java.util.concurrent.ConcurrentLinkedQueue getMessageQueue()

getEventType

public java.lang.String getEventType()

setEventType

public void setEventType(java.lang.String eventType)

getEventTypeRepository

public EventTypeRepository getEventTypeRepository()

setEventTypeRepository

public void setEventTypeRepository(EventTypeRepository eventTypeRepository)

getLastReceivedTimestamp

public long getLastReceivedTimestamp()

getSentMessageCount

public long getSentMessageCount()

getReceivedMessageCount

public long getReceivedMessageCount()

destroy

public void destroy()
             throws java.lang.Exception

Specified by:

destroy in interface DisposableBean

getState

public int getState()

setBundleContext

public void setBundleContext(BundleContext bundleContext)