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)
      Runs the task with this message installed as the current message, restoring the enclosing context afterward.

    • apply

      public <T> T apply(Function<DeserializingMessage, T> action)
      Runs an action with this message as the current message, restoring the previous context afterward. Nested actions share the enclosing batch: a nested failure propagates to its caller without completing that batch. The outermost action completes batch callbacks with its final outcome.
      Type Parameters:
      T - the result type
      Parameters:
      action - the action to run
      Returns:
      the action result

    • 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()
      Returns the lazily materialized logical message (payload, metadata, ID and timestamp). Transport fields such as source, target, request ID and index are not part of that logical message.

      Treat the serialized input as read-only after handing it to this wrapper: later mutations do not update an already materialized logical message. Use withMessage(Message) with an immutable Message replacement, or deserialize a deliberately edited serialized envelope again before dispatching it.

      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)
      Returns a message with replacement metadata while preserving the handling context, payload representation, message type, and topic.

    • withPayload

      public DeserializingMessage withPayload(Object payload)
      Returns a message with a replacement payload, preserving metadata, message ID, timestamp, message type, topic, handling context, and received transport fields (segment, index, source, target, and request ID). The replacement is serialized lazily using its own payload type and revision. Use withMessage(Message) for a new logical envelope without inherited transport fields.

    • 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. Transport fields from the original message are not inherited.

      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()
      Returns the stored message index, or a schedule deadline converted to an index for a locally constructed schedule; other local messages return null.

    • 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()
      Returns whether the message already has a deserialized representation.

    • 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()
      Returns the serialized type name, or the payload class name for a locally constructed message; returns null if no message is available.

    • getSerializedObject

      public io.fluxzero.common.api.SerializedMessage getSerializedObject()
      Returns the serialized message using this instance's serializer, serializing a local message when necessary.

    • 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)
      Replaces the serialized payload data and recreates a deserializing message with the same handling context.

    • 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)
      Wraps a batch stream for sequential handling with each message installed as the current message. Fully consuming the stream completes batch callbacks when there is no enclosing message scope; nested handling leaves completion to that scope. Short-circuiting or closing the returned stream does not complete the batch, and callers remain responsible for closing the input stream's resources.

    • 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. Nested calls leave completion to the enclosing message scope, including when an action fails.

    • whenBatchCompletes

      public static void whenBatchCompletes(io.fluxzero.common.ThrowingConsumer<Throwable> executable)
      Registers a callback for the enclosing batch's completion, or invokes it immediately outside message handling. A caught failure in nested message handling does not complete the enclosing batch or become its failure.
      Parameters:
      executable - callback receiving the error that escapes the owning scope, or null on success

    • computeForBatch

      public static <K,V> V computeForBatch(K key, BiFunction<? super K, ? super V, ? extends V> function)
      Computes a resource in the current thread's handling-batch resource map using the key and prior value; a null result removes the mapping.

    • computeForBatchIfAbsent

      public static <K,V> V computeForBatchIfAbsent(K key, Function<? super K, ? extends V> function)
      Returns the current thread's handling-batch resource or computes and stores it when absent. Unlike message-batch resources, this map is not shared with asynchronous workers through captured request context.

    • 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)
      Returns a resource from the current thread's handling-batch resource map, or null if absent.

    • getBatchResourceOrDefault

      public static <V> V getBatchResourceOrDefault(Object key, V defaultValue)
      Returns a resource from the current thread's handling-batch resource map, or the supplied default if the key is absent.

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