Package org.osgi.util.pushstream

Class SimplePushEventSourceImpl<T,U extends BlockingQueue<PushEvent<? extends T>>>

java.lang.Object

org.osgi.util.pushstream.SimplePushEventSourceImpl<T,U>

All Implemented Interfaces:

AutoCloseable, PushEventSource<T>, SimplePushEventSource<T>

class SimplePushEventSourceImpl<T,U extends BlockingQueue<PushEvent<? extends T>>>
extends Object
implements SimplePushEventSource<T>

Field Summary

Fields

Modifier and Type	Field	Description
private boolean	closed
private final List<PushEventConsumer<? super T>>	connected
private Deferred<Void>	connectPromise
private final Object	lock
private final Runnable	onClose
private final int	parallelism
private final PromiseFactory	promiseFactory
private final U	queue
private final QueuePolicy<T,U>	queuePolicy
private final PromiseFactory	sameThread
private final Semaphore	semaphore
private boolean	waitForFinishes

Constructor Summary

Constructors

Constructor	Description
SimplePushEventSourceImpl(PromiseFactory promiseFactory, QueuePolicy<T,U> queuePolicy, U queue, int parallelism, Runnable onClose)

Method Summary

Modifier and Type	Method	Description
void	close()	Close this source.
private void	close(PushEvent<T> event)
private void	closeConsumer(PushEventConsumer<? super T> pec, PushEvent<T> event)
private Promise<Void>	closedConnectPromise()
Promise<Void>	connectPromise()	This method can be used to delay event generation until an event source has connected.
private Promise<Long>	deliver(List<PushEventConsumer<? super T>> toCall, PushEvent<T> event)
private Promise<Long>	doCall(PushEvent<T> event, PushEventConsumer<? super T> pec)
private void	doSend(PushEventConsumer<? super T> pec, PushEvent<T> event)
private Promise<Long>	doSendWithBackPressure(PushEventConsumer<? super T> pec, PushEvent<T> event)
void	endOfStream()	Close this source for now, but potentially reopen it later.
private void	enqueueEvent(PushEvent<T> event)
void	error(Throwable t)	Close this source for now, but potentially reopen it later.
private void	handleReset(boolean resetWait)
boolean	isConnected()	Determine whether there are any PushEventConsumers for this PushEventSource.
AutoCloseable	open(PushEventConsumer<? super T> pec)	Open the asynchronous channel between the source and the consumer.
void	publish(T t)	Asynchronously publish an event to this stream and all connected PushEventConsumer instances.
private long	safePush(PushEventConsumer<? super T> pec, PushEvent<T> event)
private void	startWorker()

Methods inherited from class java.lang.Object

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

Field Details

lock

private final Object lock

promiseFactory

private final PromiseFactory promiseFactory

sameThread

private final PromiseFactory sameThread

queuePolicy

private final QueuePolicy<T,U extends BlockingQueue<PushEvent<? extends T>>> queuePolicy

queue

private final U extends BlockingQueue<PushEvent<? extends T>> queue

parallelism

private final int parallelism

semaphore

private final Semaphore semaphore

connected

private final List<PushEventConsumer<? super T>> connected

onClose

private final Runnable onClose

closed

private boolean closed

connectPromise

private Deferred<Void> connectPromise

waitForFinishes

private boolean waitForFinishes

Constructor Details

SimplePushEventSourceImpl

public SimplePushEventSourceImpl(PromiseFactory promiseFactory,
 QueuePolicy<T,U> queuePolicy,
 U queue,
 int parallelism,
 Runnable onClose)

Method Details

open

public AutoCloseable open(PushEventConsumer<? super T> pec)
                   throws Exception

Description copied from interface: PushEventSource

Open the asynchronous channel between the source and the consumer. The call returns an AutoCloseable. This can be closed, and should close the channel, including sending a Close event if the channel was not already closed. The returned object must be able to be closed multiple times without sending more than one Close events.

Specified by:

open in interface PushEventSource<T>

Parameters:

pec - the consumer (not null)

Returns:

a AutoCloseable that can be used to close the stream

Throws:

Exception

closeConsumer

private void closeConsumer(PushEventConsumer<? super T> pec,
 PushEvent<T> event)

doSend

private void doSend(PushEventConsumer<? super T> pec,
 PushEvent<T> event)

doSendWithBackPressure

private Promise<Long> doSendWithBackPressure(PushEventConsumer<? super T> pec,
 PushEvent<T> event)

safePush

private long safePush(PushEventConsumer<? super T> pec,
 PushEvent<T> event)

close

public void close()

Description copied from interface: SimplePushEventSource

Close this source. Calling this method indicates that there will never be any more events published by it. Calling this method sends a close event to all connected consumers. After calling this method any PushEventConsumer that tries to PushEventSource.open(PushEventConsumer) this source will immediately receive a close event, and will not see any remaining buffered events.

Specified by:

close in interface AutoCloseable

Specified by:

close in interface SimplePushEventSource<T>

close

private void close(PushEvent<T> event)

publish

public void publish(T t)

Description copied from interface: SimplePushEventSource

Asynchronously publish an event to this stream and all connected PushEventConsumer instances. When this method returns there is no guarantee that all consumers have been notified. Events published by a single thread will maintain their relative ordering, however they may be interleaved with events from other threads.

Specified by:

publish in interface SimplePushEventSource<T>

Parameters:

t -

endOfStream

public void endOfStream()

Description copied from interface: SimplePushEventSource

Close this source for now, but potentially reopen it later. Calling this method asynchronously sends a close event to all connected consumers and then disconnects them. Any events previously queued by the SimplePushEventSource.publish(Object) method will be delivered before this close event.

After calling this method any PushEventConsumer that wishes may PushEventSource.open(PushEventConsumer) this source, and will receive subsequent events.

Specified by:

endOfStream in interface SimplePushEventSource<T>

error

public void error(Throwable t)

Description copied from interface: SimplePushEventSource

Close this source for now, but potentially reopen it later. Calling this method asynchronously sends an error event to all connected consumers and then disconnects them. Any events previously queued by the SimplePushEventSource.publish(Object) method will be delivered before this error event.

After calling this method any PushEventConsumer that wishes may PushEventSource.open(PushEventConsumer) this source, and will receive subsequent events.

Specified by:

error in interface SimplePushEventSource<T>

Parameters:

t - the error

enqueueEvent

private void enqueueEvent(PushEvent<T> event)

startWorker

private void startWorker()

handleReset

private void handleReset(boolean resetWait)

deliver

private Promise<Long> deliver(List<PushEventConsumer<? super T>> toCall,
 PushEvent<T> event)

doCall

private Promise<Long> doCall(PushEvent<T> event,
 PushEventConsumer<? super T> pec)

isConnected

public boolean isConnected()

Description copied from interface: SimplePushEventSource

Determine whether there are any PushEventConsumers for this PushEventSource. This can be used to skip expensive event creation logic when there are no listeners.

Specified by:

isConnected in interface SimplePushEventSource<T>

Returns:

true if any consumers are currently connected

connectPromise

public Promise<Void> connectPromise()

Description copied from interface: SimplePushEventSource

This method can be used to delay event generation until an event source has connected. The returned promise will resolve as soon as one or more PushEventConsumer instances have opened the SimplePushEventSource.

The returned promise may already be resolved if this SimplePushEventSource already has connected consumers. If the SimplePushEventSource is closed before the returned Promise resolves then it will be failed with an IllegalStateException.

Note that the connected consumers are able to asynchronously close their connections to this SimplePushEventSource, and therefore it is possible that once the promise resolves this SimplePushEventSource may no longer be connected to any consumers.

Specified by:

connectPromise in interface SimplePushEventSource<T>

Returns:

A promise representing the connection state of this EventSource

closedConnectPromise

private Promise<Void> closedConnectPromise()