Class DeserializingMessage

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

  • Constructor Details

    • DeserializingMessage

      public DeserializingMessage(io.fluxzero.common.api.SerializedMessage message, Function<Type,Object> payload, io.fluxzero.common.MessageType messageType, String topic, Serializer serializer)

    • DeserializingMessage

      public DeserializingMessage(DeserializingObject<byte[], io.fluxzero.common.api.SerializedMessage> delegate, io.fluxzero.common.MessageType messageType, String topic, Serializer serializer)

    • DeserializingMessage

      public DeserializingMessage(@NonNull @NonNull Message message, io.fluxzero.common.MessageType messageType, Serializer serializer)

    • DeserializingMessage

      public DeserializingMessage(@NonNull @NonNull Message message, io.fluxzero.common.MessageType messageType, String topic, Serializer serializer)

    • DeserializingMessage

      protected DeserializingMessage(@NonNull @NonNull DeserializingMessage input)

  • 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 io.fluxzero.common.api.Metadata getMetadata()

    • withMetadata

      public DeserializingMessage withMetadata(io.fluxzero.common.api.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 io.fluxzero.common.api.SerializedMessage getSerializedObject()

    • withData

      public DeserializingMessage withData(io.fluxzero.common.api.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:

    • 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(io.fluxzero.common.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)

    • chunked

      default boolean chunked()

    • lastChunk

      default boolean lastChunk()

    • firstChunk

      default boolean firstChunk()