Package org.forgerock.opendj.io.rx

Class SwitchableRxSocket<M>

java.lang.Object
org.forgerock.opendj.io.rx.TransformedRxSocket<M,M>
org.forgerock.opendj.io.rx.SwitchableRxSocket<M>
Type Parameters:
M - The transport message type.
All Implemented Interfaces:
Closeable, AutoCloseable, RxSocket<M>
public final class SwitchableRxSocket<M> extends TransformedRxSocket<M,M>
A reactive socket implementation which delegates to a replaceable delegate reactive socket. This class may be used for implementing features like StartTLS or SASL QOS where the application layer negotiates additional security layers.

  • Constructor Details

    • SwitchableRxSocket

      public SwitchableRxSocket(RxSocket<M> socket)
      Returns a new switchable socket delegating to the provided downstream socket.
      Parameters:
      socket - The downstream socket.

  • Method Details

    • suspendReads

      public void suspendReads()
      Prevents further attempts to read from the socket until either resumeReadsAfterSwitchingSocket(org.forgerock.opendj.io.RxSocket<M>) or resumeReads() are invoked. It is safe to replace the socket while reads are suspended. This method must be invoked by the reader thread.

    • resumeReadsAfterSwitchingSocket

      public void resumeReadsAfterSwitchingSocket(RxSocket<M> newSocket)
      Replaces the active socket with a new socket and then resumes reading. This method is typically invoked by a writer thread after it has successfully negotiated a new network layer and only after suspendReads() was previously invoked by the reader thread.
      Parameters:
      newSocket - The socket which will become the active socket.

    • resumeReads

      public void resumeReads()
      Resumes reading without replacing the active socket. This method is typically invoked by a writer thread after it has after it has unsuccessfully negotiated a new network layer and only after suspendReads() was previously invoked by the reader thread.

    • read

      public Single<M> read(Supplier<M> bufferSupplier)
      Description copied from interface: RxSocket
      Returns a "cold" Single which will read the next message from the network each time it is subscribed. Read operations cannot be cancelled: cancelling/disposing a subscription has no effect, and will not close the underlying socket.

      Socket implementations must buffer incoming messages until they are read. Note that invoking this method multiple times returns the same Single instance. The returned Single will not allow multiple concurrent subscriptions as this will result in non-deterministic behavior. Attempts to have multiple concurrent subscriptions will result in an IllegalStateException.

      Subscribers will be notified when the socket is closed through SingleObserver.onError(Throwable) with the corresponding IOException providing the reason for the disconnection. In particular, if the socket is closed locally via RxSocket.close() then it will be notified with RxSocket.LOCAL_CLOSE, if it is closed locally via RxSocket.closeWithReason(java.lang.Throwable) then it will be notified using the provided exception and finally, if it is closed remotely by the peer, then it will be notified using RxSocket.REMOTE_CLOSE. Low-level network transports should return a SocketTimeoutException for communicating any network timeouts. Application layers may return exceptions which are more closely aligned to their APIs.

      Specified by:
      read in interface RxSocket<M>
      Specified by:
      read in class TransformedRxSocket<M,M>
      Parameters:
      bufferSupplier - An optional buffer supplier which may be used by the transport to reduce copying between transport layers. Use null when the transport does not support buffer suppliers or if the transport should allocate its own buffer. Note that the buffer supplier may be called multiple times in order to complete a single read.
      Returns:
      A "cold" Single which will be signalled once the read operation completes or fails.

    • write

      public Completable write(M message)
      Description copied from interface: RxSocket
      Returns a "cold" Completable which will write the provided message to the network each time it is subscribed. Write operations cannot be cancelled reliably, as a consequence cancelling/disposing a subscription has no effect.

      Subscribers will be notified when the socket is closed through CompletableObserver.onError(Throwable) with the corresponding IOException providing the reason for the disconnection. In particular, if the socket is closed locally via RxSocket.close() then it will be notified with RxSocket.LOCAL_CLOSE, if it is closed locally via RxSocket.closeWithReason(java.lang.Throwable) then it will be notified using the provided exception and finally, if it is closed remotely by the peer, then it will be notified using RxSocket.REMOTE_CLOSE. Low-level network transports should return a SocketTimeoutException for communicating any network timeouts. Application layers may return exceptions which are more closely aligned to their APIs.

      This method is thread-safe, atomic and ordered with respect to other write attempts. In particular, consecutive writes will be performed in the order in which they are subscribed.

      Specified by:
      write in interface RxSocket<M>
      Specified by:
      write in class TransformedRxSocket<M,M>
      Parameters:
      message - The message to be written to the network.
      Returns:
      A "cold" Completable which will be signalled once the write operation completes or fails.

    • getSocket

      public RxSocket<M> getSocket()
      Returns the active socket.
      Returns:
      The active socket.

    • getSocket

      public <S extends RxSocket<M>> S getSocket(Class<S> clazz)
      Returns the active socket if it has the provided class or null otherwise.
      Type Parameters:
      S - The expected class of the active socket.
      Parameters:
      clazz - The expected class of the active socket.
      Returns:
      The active socket or null.

    • setSocket

      public void setSocket(RxSocket<M> newSocket)
      Sets the active socket. This method may be used in order to directly replace the active socket without any coordination with the reader. It is typically safe to do this on the client side once the final protocol handshake (e.g. StartTLS response) is received.
      Parameters:
      newSocket - The new active socket.