Top Description Inners Fields Constructors Methods
io.rsocket.resume

public Class InMemoryResumableFramesStore

extends Flux<ByteBuf>
implements ResumableFramesStore, Subscription
Class Inheritance
All Implemented Interfaces
org.reactivestreams.Subscription, io.rsocket.resume.ResumableFramesStore, io.rsocket.Closeable, reactor.core.Disposable
Static Imports
io.rsocket.resume.ResumableDuplexConnection.isResumableFrame
writes - n (where n is frequent, primary operation) reads - m (where m == KeepAliveFrequency) skip - k -> 0 (where k is the rare operation which happens after disconnection

Nested and Inner Type Summary

Modifier and TypeClass and Description
pack-priv static class
InMemoryResumableFramesStore.FramesSubscriber

Field Summary

Modifier and TypeField and Description
pack-priv CoreSubscriber<? super ByteBuf>
actual

pack-priv final Queue<ByteBuf>
cachedFrames

pack-priv final int
cacheLimit

pack-priv int
cacheSize

pack-priv static final long
CONNECTED_FLAG

Flag which indicates that InMemoryResumableFramesStore has active frames consumer

pack-priv final Sinks.Empty<Void>
disposed

pack-priv static final long
DISPOSED_FLAG

Flag which indicates that InMemoryResumableFramesStore is terminated via the InMemoryResumableFramesStore#dispose() method

pack-priv static final long
FINALIZED_FLAG

Flag which indicates that InMemoryResumableFramesStore is finalized and all related stores are cleaned

pack-priv static final AtomicLongFieldUpdater<InMemoryResumableFramesStore>
FIRST_AVAILABLE_FRAME_POSITION

pack-priv volatile long
firstAvailableFramePosition

private InMemoryResumableFramesStore.FramesSubscriber
framesSubscriber

pack-priv static final long
HAS_FRAME_FLAG

Flag which indicates that there are some frames stored in the io.rsocket.internal.UnboundedProcessor which has to be cached and sent to the remote party

pack-priv static final AtomicLongFieldUpdater<InMemoryResumableFramesStore>
IMPLIED_POSITION

pack-priv volatile long
impliedPosition

private static final Logger
logger

pack-priv static final long
MAX_WORK_IN_PROGRESS

Flag which indicates that InMemoryResumableFramesStore#drain(long) has an actor which is currently progressing on the work.

pack-priv static final long
PENDING_CONNECTION_FLAG

Flag which indicates that InMemoryResumableFramesStore has no active frames consumer but there is a one pending

pack-priv CoreSubscriber<? super ByteBuf>
pendingActual

pack-priv static final long
REMOTE_IMPLIED_POSITION_CHANGED_FLAG

Flag which indicates that there are some received implied position changes from the remote party

pack-priv long
remoteImpliedPosition

pack-priv final String
session

pack-priv final String
side

pack-priv volatile long
state

pack-priv static final AtomicLongFieldUpdater<InMemoryResumableFramesStore>
STATE

pack-priv Throwable
terminal

pack-priv static final long
TERMINATED_FLAG

Flag which indicates that InMemoryResumableFramesStore is terminated via the FramesSubscriber#onComplete() or FramesSubscriber#onError(Throwable) ()} methods

Constructor Summary

AccessConstructor and Description
public
InMemoryResumableFramesStore(String side, ByteBuf session, int cacheSizeBytes)

Method Summary

Modifier and TypeMethod and Description
public void
cancel()

Implements org.reactivestreams.Subscription.cancel.

Request the Publisher to stop sending data and clean up resources.
pack-priv static void
clearAndFinalize(InMemoryResumableFramesStore store)

pack-priv void
clearCache()

public void
dispose()

Implements reactor.core.Disposable.dispose.

Cancel or dispose the underlying task or resource.
pack-priv void
drain(long expectedState)

pack-priv static int
dropFramesFromCache(long toRemoveBytes, Queue<ByteBuf> cache)

public long
frameImpliedPosition()

Implements io.rsocket.resume.ResumableFramesStore.frameImpliedPosition.

public long
framePosition()

Implements io.rsocket.resume.ResumableFramesStore.framePosition.

pack-priv void
handleConnectionFrame(ByteBuf frame)

pack-priv void
handleDisposed()

pack-priv void
handleFrame(ByteBuf frame)

pack-priv long
handlePendingConnection(long expectedState, Queue<ByteBuf> cachedFrames)

pack-priv void
handlePendingFrames(Fuseable.QueueSubscription<ByteBuf> qs)

pack-priv long
handlePendingRemoteImpliedPositionChanges(long expectedState, Queue<ByteBuf> cachedFrames)

pack-priv void
handleResumableFrame(ByteBuf frame)

pack-priv void
handleTerminated(Fuseable.QueueSubscription<ByteBuf> qs, Throwable t)

pack-priv static boolean
hasFrames(long state)

pack-priv static boolean
hasPendingConnection(long state)

pack-priv static boolean
hasRemoteImpliedPositionChanged(long state)

pack-priv static boolean
isConnected(long state)

public boolean
isDisposed()

Overrides default reactor.core.Disposable.isDisposed.

Optionally return true when the resource or task is disposed.
pack-priv static boolean
isDisposed(long state)

pack-priv static boolean
isFinalized(long state)

pack-priv static boolean
isTerminated(long state)

pack-priv static boolean
isWorkInProgress(long state)

pack-priv static long
markConnected(InMemoryResumableFramesStore store, long expectedState)

pack-priv static long
markDisconnected(InMemoryResumableFramesStore store)

pack-priv static long
markDisposed(InMemoryResumableFramesStore store)

pack-priv static long
markFrameAdded(InMemoryResumableFramesStore store)

pack-priv static long
markPendingConnection(InMemoryResumableFramesStore store)

pack-priv static long
markRemoteImpliedPositionChanged(InMemoryResumableFramesStore store)

pack-priv static long
markTerminated(InMemoryResumableFramesStore store)

pack-priv static long
markWorkDone(InMemoryResumableFramesStore store, long expectedState)

public Mono<Void>
onClose()

Implements io.rsocket.Closeable.onClose.

Returns a Mono that terminates when the instance is terminated by any reason.
pack-priv void
pauseImplied()

public void
releaseFrames(long remoteImpliedPos)

Implements io.rsocket.resume.ResumableFramesStore.releaseFrames.

Release frames from tail of the store up to remote implied position
public void
request(long
the strictly positive number of elements to requests to the upstream Publisher
n)

Implements org.reactivestreams.Subscription.request.

No events will be sent by a Publisher until demand is signaled via this method.
public boolean
resumableFrameReceived(ByteBuf frame)

Implements io.rsocket.resume.ResumableFramesStore.resumableFrameReceived.

Received resumable frame as defined by RSocket protocol.
pack-priv void
resumeImplied()

public Flux<ByteBuf>
resumeStream()

Implements io.rsocket.resume.ResumableFramesStore.resumeStream.

public Mono<Void>
saveFrames(Flux<ByteBuf>
Flux of resumable frames
frames)

Implements io.rsocket.resume.ResumableFramesStore.saveFrames.

Save resumable frames for potential resumption
public void
subscribe(CoreSubscriber<? super ByteBuf>
the Subscriber interested into the published sequence
actual)

Implements abstract reactor.core.publisher.Flux.subscribe.

Implements reactor.core.CorePublisher.subscribe.

An internal Publisher#subscribe(Subscriber) that will bypass Hooks#onLastOperator(Function) pointcut.
Inherited from reactor.core.publisher.Flux:
allanyasblockFirstblockFirstblockLastblockLastbufferbufferbufferbufferbufferbufferbufferbufferbufferbufferbufferbufferTimeoutbufferTimeoutbufferTimeoutbufferTimeoutbufferTimeoutbufferTimeoutbufferTimeoutbufferTimeoutbufferUntilbufferUntilbufferUntilChangedbufferUntilChangedbufferUntilChangedbufferWhenbufferWhenbufferWhilecachecachecachecachecachecachecancelOncastcheckpointcheckpointcheckpointcollectcollectcollectListcollectMapcollectMapcollectMapcollectMultimapcollectMultimapcollectMultimapcollectSortedListcollectSortedListcombineLatestcombineLatestcombineLatestcombineLatestcombineLatestcombineLatestcombineLatestcombineLatestcombineLatestconcatconcatconcatconcatconcatDelayErrorconcatDelayErrorconcatDelayErrorconcatDelayErrorconcatMapconcatMapconcatMapDelayErrorconcatMapDelayErrorconcatMapDelayErrorconcatMapIterableconcatMapIterableconcatWithconcatWithValuescontextCapturecontextWritecontextWritecountcreatecreatedefaultIfEmptydeferdeferContextualdelayElementsdelayElementsdelaySequencedelaySequencedelaySubscriptiondelaySubscriptiondelaySubscriptiondelayUntildematerializedistinctdistinctdistinctdistinctdistinctUntilChangeddistinctUntilChangeddistinctUntilChangeddoAfterTerminatedoFinallydoFirstdoOnCanceldoOnCompletedoOnDiscarddoOnEachdoOnErrordoOnErrordoOnErrordoOnNextdoOnRequestdoOnSubscribedoOnTerminateelapsedelapsedelementAtelementAtemptyerrorerrorerrorexpandexpandexpandDeepexpandDeepfilterfilterWhenfilterWhenfirstfirstfirstWithSignalfirstWithSignalfirstWithValuefirstWithValueflatMapflatMapflatMapflatMapflatMapDelayErrorflatMapIterableflatMapIterableflatMapSequentialflatMapSequentialflatMapSequentialflatMapSequentialDelayErrorfromfromArrayfromIterablefromStreamfromStreamgenerategenerategenerategetPrefetchgroupBygroupBygroupBygroupBygroupJoinhandlehasElementhasElementshideignoreElementsindexindexintervalintervalintervalintervaljoinjustjustlastlastlimitRatelimitRatelimitRequestloglogloglogloglogmapmapNotNullmaterializemergemergemergemergemergemergemergeComparingmergeComparingmergeComparingmergeComparingDelayErrormergeComparingWithmergeDelayErrormergeOrderedmergeOrderedmergeOrderedmergeOrderedWithmergePrioritymergePrioritymergePrioritymergePriorityDelayErrormergeSequentialmergeSequentialmergeSequentialmergeSequentialmergeSequentialmergeSequentialmergeSequentialDelayErrormergeSequentialDelayErrormergeSequentialDelayErrormergeWithmetricsnamenevernextofTypeonAssemblyonAssemblyonBackpressureBufferonBackpressureBufferonBackpressureBufferonBackpressureBufferonBackpressureBufferonBackpressureBufferonBackpressureBufferonBackpressureDroponBackpressureDroponBackpressureErroronBackpressureLatestonErrorCompleteonErrorCompleteonErrorCompleteonErrorContinueonErrorContinueonErrorContinueonErrorMaponErrorMaponErrorMaponErrorResumeonErrorResumeonErrorResumeonErrorReturnonErrorReturnonErrorReturnonErrorStoponTerminateDetachorparallelparallelparallelpublishpublishpublishpublishpublishNextpublishOnpublishOnpublishOnpushpushrangereducereducereduceWithrepeatrepeatrepeatrepeatrepeatWhenreplayreplayreplayreplayreplayreplayretryretryretryWhensamplesamplesampleFirstsampleFirstsampleTimeoutsampleTimeoutscanscanscanWithshareshareNextsinglesinglesingleOrEmptyskipskipskipskipLastskipUntilskipUntilOtherskipWhilesortsortstartWithstartWithstartWithsubscribesubscribesubscribesubscribesubscribesubscribesubscribesubscribeOnsubscribeOnsubscribeWithswitchIfEmptyswitchMapswitchMapswitchOnFirstswitchOnFirstswitchOnNextswitchOnNexttagtaketaketaketaketakeLasttakeUntiltakeUntilOthertakeWhiletaptaptapthenthenthenEmptythenManytimedtimedtimeouttimeouttimeouttimeouttimeouttimeouttimeouttimestamptimestamptoIterabletoIterabletoIterabletoStreamtoStreamtoStringtransformtransformDeferredtransformDeferredContextualusingusingusingWhenusingWhenwindowwindowwindowwindowwindowwindowwindowwindowTimeoutwindowTimeoutwindowTimeoutwindowTimeoutwindowUntilwindowUntilwindowUntilwindowUntilChangedwindowUntilChangedwindowUntilChangedwindowWhenwindowWhilewindowWhilewithLatestFromzipzipzipzipzipzipzipzipzipzipzipzipzipzipWithzipWithzipWithzipWithzipWithIterablezipWithIterable