Class DeserializingMessage

java.lang.Object
io.fluxzero.sdk.common.serialization.DeserializingMessage
All Implemented Interfaces:
HasMetadata, HasMessage
Direct Known Subclasses:
ChunkedDeserializingMessage, DefaultEntityHelper.DeserializingMessageWithEntity, StatefulHandler.StatefulHandlerSearchMessage, StatefulHandler.StatefulMemberMessage
public class DeserializingMessage extends Object implements HasMessage
Wrapper for a Message that supports lazy deserialization, context caching, type adaptation, and batch-level execution utilities.

DeserializingMessage combines a SerializedMessage with deserialization and routing logic while maintaining the original message context (type, topic, metadata, and payload).

Key Features

  • Supports on-demand deserialization of a Message
  • Provides thread-local access to the message that is currently being handled
  • Allows attaching resources to the current message or message batch/li>
See Also:

  • Field Details

    • messageFormatter

      public static MessageFormatter messageFormatter
      The formatter used to produce a human-readable representation of this message, primarily for logging or debugging. By default, this uses MessageFormatter.DEFAULT.

      In advanced scenarios, users may replace this field with a custom MessageFormatter implementation to modify how deserializing messages are rendered (e.g., to include metadata or correlation IDs).

  • Constructor Details

  • Method Details

    • run

      public void run(Consumer<DeserializingMessage> task)

    • apply

      public <T> T apply(Function<DeserializingMessage, T> action)

    • captureContext

      public ThreadLocalContext.Snapshot captureContext()
      Captures the complete request context with this message bound as the current message.

      This also works for framework paths that prepare an asynchronous callback outside a surrounding apply(Function) call. The context that was active before capturing is restored immediately.

    • toMessage

      public Message toMessage()
      Description copied from interface: HasMessage
      Returns the underlying Message representation of this object.
      Specified by:
      toMessage in interface HasMessage
      Returns:
      the Message backing this instance

    • getMetadata

      public Metadata getMetadata()
      Description copied from interface: HasMetadata
      Returns the Metadata associated with this object.
      Specified by:
      getMetadata in interface HasMetadata
      Returns:
      metadata attached to this instance; never null

    • withMetadata

      public DeserializingMessage withMetadata(Metadata metadata)

    • withPayload

      public DeserializingMessage withPayload(Object payload)

    • getMessageId

      public String getMessageId()
      Description copied from interface: HasMessage
      Returns the unique ID of the underlying message.
      Specified by:
      getMessageId in interface HasMessage
      Returns:
      the message ID

    • getIndex

      public Long getIndex()

    • getTimestamp

      public Instant getTimestamp()
      Description copied from interface: HasMessage
      Returns the timestamp at which the message was created or published.
      Specified by:
      getTimestamp in interface HasMessage
      Returns:
      the message timestamp

    • isDeserialized

      public boolean isDeserialized()

    • getPayload

      public <V> V getPayload()
      Description copied from interface: HasMessage
      Retrieves the message payload, deserializing if necessary, cast to the expected type.

      By default, this delegates to toMessage().getPayload().

      Specified by:
      getPayload in interface HasMessage
      Type Parameters:
      V - the expected payload type
      Returns:
      the deserialized payload

    • getPayloadAs

      public <R> R getPayloadAs(Type type)
      Description copied from interface: HasMessage
      Retrieves the message payload, deserializing if necessary and optionally converted to the given type.

      By default, this performs a conversion of the payload using JsonUtils.

      Specified by:
      getPayloadAs in interface HasMessage
      Type Parameters:
      R - the expected payload type
      Returns:
      the payload converted to the given type

    • getPayloadClass

      public Class<?> getPayloadClass()
      Description copied from interface: HasMessage
      Returns the runtime class of the payload object, or Void.class if the payload is null.
      Specified by:
      getPayloadClass in interface HasMessage
      Returns:
      the payload's class

    • getType

      public String getType()

    • getSerializedObject

      public SerializedMessage getSerializedObject()

    • withData

      public DeserializingMessage withData(Data<byte[]> data)

    • computeContextIfAbsent

      public <T> T computeContextIfAbsent(Class<T> contextKey, Function<DeserializingMessage, ? extends T> provider)
      Returns a message-scoped context value, computing and storing it when it is absent.
      Parameters:
      contextKey - the type key for the context value
      provider - the provider used to compute the value when absent
      Returns:
      the existing or computed context value

    • getContext

      public <T> Optional<T> getContext(Class<T> contextKey)
      Returns a message-scoped context value.
      Parameters:
      contextKey - the type key for the context value
      Returns:
      the context value, or Optional.empty() when absent

    • putContext

      public <T> DeserializingMessage putContext(Class<T> contextKey, T value)
      Stores a message-scoped context value.
      Parameters:
      contextKey - the type key for the context value
      value - the context value to store
      Returns:
      this message

    • getCurrent

      public static DeserializingMessage getCurrent()
      Returns the current DeserializingMessage being processed in this thread, or null if none is set.

      This method provides direct (nullable) access to the thread-local message context. Prefer getOptionally() when you want to safely handle absence of context.

      Note: This method should typically be called only inside handler code or interceptors where a DeserializingMessage is known to be active.

      Returns:
      the current message or null if no message is being processed
      See Also:

    • hasThreadLocalContext

      public static boolean hasThreadLocalContext()
      Returns whether this thread already has a regular deserializing-message context.

      This method does not count a payload-first local execution that can be exposed through getCurrent(). It is intended for infrastructure that must avoid nesting a regular message context.

      Returns:
      true if a regular message context is bound to the current thread

    • getOptionally

      public static Optional<DeserializingMessage> getOptionally()
      Returns the current DeserializingMessage being processed in this thread, if available.

      This method is safe to call in any thread and will return Optional.empty() if no message is currently being handled. It is particularly useful for utility classes or exception handlers that want to conditionally access message metadata.

      Example

      Optional<DeserializingMessage> message = DeserializingMessage.getOptionally();
message.map(DeserializingMessage::getPayloadType)
       .ifPresent(type -> log.debug("Handling message of type {}", type));
      Returns:
      an Optional containing the current message or empty if none is set
      See Also:

    • toString

      public String toString()
      Overrides:
      toString in class Object

    • handleBatch

      public static Stream<DeserializingMessage> handleBatch(Stream<DeserializingMessage> batch)

    • forEachInBatch

      public static void forEachInBatch(Iterable<DeserializingMessage> batch, Consumer<? super DeserializingMessage> action)
      Processes messages while exposing each message through getCurrent() and completing batch-scoped callbacks after the last message. This is the lower-allocation counterpart to handleBatch(Stream) for callers that already have an iterable batch.

    • whenBatchCompletes

      public static void whenBatchCompletes(ThrowingConsumer<Throwable> executable)

    • computeForBatch

      public static <K,V> V computeForBatch(K key, BiFunction<? super K, ? super V, ? extends V> function)

    • computeForBatchIfAbsent

      public static <K,V> V computeForBatchIfAbsent(K key, Function<? super K, ? extends V> function)

    • getBatchResource

      public static <V> V getBatchResource(Object key)

    • getBatchResourceOrDefault

      public static <V> V getBatchResourceOrDefault(Object key, V defaultValue)

    • completeLocalBatch

      public static void completeLocalBatch(Throwable error)
      Runs and clears batch-completion callbacks and resources after a payload-first local invocation.

      This method is intended for Fluxzero handling infrastructure. Applications normally register callbacks with whenBatchCompletes(ThrowingConsumer) instead of completing a batch directly.

      Parameters:
      error - the error that ended handling, or null after successful handling