Class DefaultRequestHandler

java.lang.Object
io.fluxzero.sdk.common.AbstractNamespaced<RequestHandler>
io.fluxzero.sdk.publishing.DefaultRequestHandler
All Implemented Interfaces:
Namespaced<RequestHandler>, RequestHandler, AutoCloseable
public class DefaultRequestHandler extends AbstractNamespaced<RequestHandler> implements RequestHandler
Default implementation of the RequestHandler interface.

This handler supports both single and batch request dispatching, tracking responses using an internal ConcurrentHashMap keyed by requestId. When a request is sent, the handler subscribes to a corresponding result log (e.g., result or web response) via a TrackingClient, which listens for responses targeted at this client only.

Each request is assigned a unique requestId and tagged with the client's source identifier. When a response with a matching requestId is received, the corresponding CompletableFuture is completed.

If no response is received within the configured timeout (default: 200 seconds), the future is completed exceptionally.

This request handle supports chunked responses. Request senders that can deal with chunked responses should use sendRequest(SerializedMessage, Consumer, Duration, Consumer)}. If a chunked response is received, but the request sender expected a single response, the intermediate responses are aggregated before completing the request.

Features:

  • Supports both single and batch request dispatching.
  • Tracks responses via the configured MessageType and filters using filterMessageTarget = true.
  • Ensures startup of the underlying result tracker on first request dispatch.
  • Cleans up subscriptions and pending futures on close().
See Also:

  • Constructor Details

    • DefaultRequestHandler

      public DefaultRequestHandler(Client client, io.fluxzero.common.MessageType resultType, Duration timeout, String responseConsumerName)
      Constructs a DefaultRequestHandler with the specified client, message type, timeout, and response consumer name. This constructor creates an internal worker pool for handling requests and responses.
      Parameters:
      client - the client responsible for sending and receiving messages
      resultType - the type of message expected as a result
      timeout - the maximum duration to wait for a response
      responseConsumerName - the name of the consumer responsible for handling the response

    • DefaultRequestHandler

      public DefaultRequestHandler(Client client, io.fluxzero.common.MessageType resultType)
      Constructs a DefaultRequestHandler with the specified client and message type, and a default timeout of 200 seconds. This constructor creates an internal worker pool for handling requests and responses.

      Uses a default name for the result consumer based on the application name.

      Parameters:
      client - the client responsible for sending and receiving messages
      resultType - the type of message expected as a result

  • Method Details

    • createForNamespace

      protected RequestHandler createForNamespace(String namespace)
      Specified by:
      createForNamespace in class AbstractNamespaced<RequestHandler>

    • sendRequest

      public CompletableFuture<io.fluxzero.common.api.SerializedMessage> sendRequest(io.fluxzero.common.api.SerializedMessage request, Consumer<io.fluxzero.common.api.SerializedMessage> requestSender, Duration timeout)
      Sends a request and processes the response, combining intermediate responses (if any) with the final response data. This method ensures intermediate results are aggregated and included in the final output.
      Specified by:
      sendRequest in interface RequestHandler
      Parameters:
      request - The request message to be sent.
      requestSender - A callback used to dispatch the request.
      timeout - The timeout for this request; null uses the handler default and a negative value disables the timeout.
      Returns:
      A CompletableFuture that completes with the response or fails on timeout.

    • sendRequest

      public CompletableFuture<io.fluxzero.common.api.SerializedMessage> sendRequest(io.fluxzero.common.api.SerializedMessage request, Consumer<io.fluxzero.common.api.SerializedMessage> requestSender, Duration timeout, Consumer<io.fluxzero.common.api.SerializedMessage> intermediateCallback)
      Specified by:
      sendRequest in interface RequestHandler

    • sendRequests

      public List<CompletableFuture<io.fluxzero.common.api.SerializedMessage>> sendRequests(List<io.fluxzero.common.api.SerializedMessage> requests, Consumer<List<io.fluxzero.common.api.SerializedMessage>> requestSender)
      Description copied from interface: RequestHandler
      Sends multiple requests and returns a list of futures for their corresponding responses.

      Each request is assigned a unique requestId and dispatched using the given sender. The returned list preserves the order of the input requests.

      Specified by:
      sendRequests in interface RequestHandler
      Parameters:
      requests - The requests to send.
      requestSender - A callback used to dispatch the requests (e.g. batch publisher).
      Returns:
      A list of CompletableFuture instances, one for each request.

    • sendRequests

      public List<CompletableFuture<io.fluxzero.common.api.SerializedMessage>> sendRequests(List<io.fluxzero.common.api.SerializedMessage> requests, Consumer<List<io.fluxzero.common.api.SerializedMessage>> requestSender, Duration timeout)
      Description copied from interface: RequestHandler
      Sends multiple requests with a custom timeout and returns a list of futures for their responses.

      A non-negative timeout is also written to RequestHandler.REQUEST_TIMEOUT_METADATA_KEY on each request so downstream request handlers can recognize stale requests during tracking. A null timeout falls back to this handler's configured default.

      Specified by:
      sendRequests in interface RequestHandler
      Parameters:
      requests - The requests to send.
      requestSender - A callback used to dispatch the requests.
      timeout - The timeout to apply per request; null uses the handler default and a negative value disables the timeout.
      Returns:
      A list of CompletableFuture instances, one for each request.

    • prepareRequest

      protected CompletableFuture<io.fluxzero.common.api.SerializedMessage> prepareRequest(io.fluxzero.common.api.SerializedMessage request, Duration timeout, Consumer<io.fluxzero.common.api.SerializedMessage> intermediateCallback)

    • ensureStarted

      protected void ensureStarted()

    • handleResults

      protected void handleResults(List<io.fluxzero.common.api.SerializedMessage> messages)

    • completeRequestExceptionally

      protected boolean completeRequestExceptionally(int requestId, Throwable error)
      Completes a pending request exceptionally and removes its response callback.
      Parameters:
      requestId - the request id assigned by prepareRequest(SerializedMessage, Duration, Consumer)
      error - the error that should complete the pending request
      Returns:
      true when a pending request was found and completed

    • close

      public void close()
      Description copied from interface: RequestHandler
      Releases all resources associated with this handler.

      This typically shuts down any underlying TrackingClient subscriptions, and may cancel or complete any outstanding requests.

      Specified by:
      close in interface AutoCloseable
      Specified by:
      close in interface RequestHandler
      Overrides:
      close in class AbstractNamespaced<RequestHandler>