Class DeserializingMessage

java.lang.Object
io.fluxzero.sdk.common.serialization.DeserializingMessage
All Implemented Interfaces:
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

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

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

    • containsMetadata

      public boolean containsMetadata(String key)
      Checks one metadata key without forcing an opaque serialized message to materialize its metadata wrapper.

    • getMetadataValue

      public String getMetadataValue(String key)
      Reads one metadata value without forcing an opaque serialized message to materialize its metadata wrapper.

    • withMetadata

      public DeserializingMessage withMetadata(io.fluxzero.common.api.Metadata metadata)

    • withPayload

      public DeserializingMessage withPayload(Object payload)

    • withRestoredPayload

      public DeserializingMessage withRestoredPayload(Object payload)
      Exposes a restored, same-type payload to handling code while retaining this message's serialized envelope.

      This is intended for values restored from private storage, not for changing the event. Metadata-only copies retain the original data, revision and index. A subsequent withPayload(Object) or withMessage(Message) is a real replacement and uses ordinary serialization again.

      Parameters:
      payload - the restored logical payload
      Returns:
      a view with the restored payload and the unchanged serialized source

    • withMessage

      public DeserializingMessage withMessage(Message message)
      Replaces the complete logical message while retaining this wrapper's type, topic, serializer, and context.

      Unlike withPayload(Object), this method deliberately retains the supplied message identity. It is used when one handled message expands into several independently publishable model updates.

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

    • getSerializedObject

      public io.fluxzero.common.api.SerializedMessage getSerializedObject(Serializer fallbackSerializer)
      Returns the existing serialized representation or lazily creates one using the supplied fallback serializer. The fallback is only used for programmatically created messages that were not constructed with a serializer.

    • 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

    • withoutContext

      public DeserializingMessage withoutContext(Class<?> contextKey)
      Returns this message after removing one framework context value using copy-on-write semantics. Other messages derived from the same source retain their context.

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

    • computeForMessageBatchIfAbsent

      public static <K,V> V computeForMessageBatchIfAbsent(K key, Function<? super K, ? extends V> function)
      Returns one resource shared by the complete current message batch, creating it when necessary.

      Unlike computeForBatchIfAbsent(Object, Function), this resource participates in captured request context and is therefore visible to asynchronous handler workers. The mapping function and returned value must consequently be safe for concurrent use. Outside message-batch handling this method returns null.

    • getMessageBatchResource

      public static <V> V getMessageBatchResource(Object key)
      Returns a previously created asynchronous message-batch resource, or null outside that batch.

    • getMessageBatchIndex

      public static int getMessageBatchIndex()
      Returns the zero-based position of the current message in its tracking batch, or -1 outside a batch.

    • getMessageBatchSegment

      public static int getMessageBatchSegment()
      Returns the routing segment of the current message-batch position, or -1 outside a message batch.

    • getMessageBatchSize

      public static int getMessageBatchSize()
      Returns the known message count of the current batch, or -1 when its source did not expose a size.

    • 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

    • chunked

      default boolean chunked()

    • lastChunk

      default boolean lastChunk()

    • firstChunk

      default boolean firstChunk()