Class DeserializingMessage

java.lang.Object
io.fluxzero.sdk.common.serialization.DeserializingMessage
All Implemented Interfaces:
HasMetadata, HasMessage
Direct Known Subclasses:
DefaultEntityHelper.DeserializingMessageWithEntity
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)

    • 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)

    • 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:

    • 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)

    • 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)