org.axonframework.kafka.eventhandling

Class HeaderUtils

java.lang.Object

org.axonframework.kafka.eventhandling.HeaderUtils

public class HeaderUtils
extends Object

Utility for dealing with Headers. Mostly for internal use.

Since:

3.0

Author:

Nakul Mishra

Method Summary

Modifier and Type	Method and Description
static void	addHeader(org.apache.kafka.common.header.Headers target, String key, Object value) Creates a new RecordHeader based on key and value and adds it to target.
static Long	asLong(byte[] value) Converts bytes to long.
static String	asString(byte[] value) Converts bytes to String.
static BiFunction<String,Object,org.apache.kafka.common.header.internals.RecordHeader>	byteMapper() Handles value conversion to bytes.
static Map<String,Object>	extractAxonMetadata(org.apache.kafka.common.header.Headers target) Extract axon metadata(if any) attached with Headers.
static String	extractKey(String metadataKey) Extracts actual key name used to send Axon metadata.
static String	generateMetadataKey(String key) Generates key, used to identify Axon metadata in RecordHeader.
static Set<String>	keys(org.apache.kafka.common.header.Headers headers) Extract keys from Headers.
static byte[]	toBytes(Number value) Converts primitive arithmetic types to byte array.
static org.apache.kafka.common.header.Headers	toHeaders(EventMessage<?> eventMessage, SerializedObject<byte[]> serializedObject, BiFunction<String,Object,org.apache.kafka.common.header.internals.RecordHeader> headerValueMapper) Generates Headers based on EventMessage and SerializedObject.
static byte[]	value(org.apache.kafka.common.header.Headers headers, String key) Return the value stored under a given key inside the Headers.
static Long	valueAsLong(org.apache.kafka.common.header.Headers headers, String key) Return Long representation of the value stored under a given key inside the Headers.
static String	valueAsString(org.apache.kafka.common.header.Headers headers, String key) Return String representation of the value stored under a given key inside the Headers.
static String	valueAsString(org.apache.kafka.common.header.Headers headers, String key, String defaultValue) Return String representation of the value stored under a given key inside the Headers.

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Method Detail

asLong

public static Long asLong(byte[] value)

Converts bytes to long.

Parameters:

value - representing long.

Returns:

the long.

valueAsLong

public static Long valueAsLong(org.apache.kafka.common.header.Headers headers,
                               String key)

Return Long representation of the value stored under a given key inside the Headers. In case of missing entry null is returned.

Parameters:

headers - the Kafka headers.

key - the expected key.

Returns:

the value.

asString

public static String asString(byte[] value)

Converts bytes to String.

Parameters:

value - bytes representing String.

Returns:

the String.

valueAsString

public static String valueAsString(org.apache.kafka.common.header.Headers headers,
                                   String key)

Return String representation of the value stored under a given key inside the Headers. In case of missing entry null is returned.

Parameters:

headers - the Kafka headers.

key - the expected key.

Returns:

the value.

valueAsString

public static String valueAsString(org.apache.kafka.common.header.Headers headers,
                                   String key,
                                   String defaultValue)

Return String representation of the value stored under a given key inside the Headers. In case of missing entry defaultValue is returned.

Parameters:

headers - the Kafka headers.

key - the expected key.

defaultValue - the value when key is missing.

Returns:

the value

value

public static byte[] value(org.apache.kafka.common.header.Headers headers,
                           String key)

Return the value stored under a given key inside the Headers. In case of missing entry null is returned.

Parameters:

headers - the Kafka headers.

key - the expected key.

Returns:

the value.

toBytes

public static byte[] toBytes(Number value)

Converts primitive arithmetic types to byte array.

Parameters:

value - the number.

Returns:

the bytes.

addHeader

public static void addHeader(org.apache.kafka.common.header.Headers target,
                             String key,
                             Object value)

Creates a new RecordHeader based on key and value and adds it to target. The value is converted to bytes and follows this logic:

• Instant - calls Instant.toEpochMilli()
• Number - calls toBytes(java.lang.Number)
• String/custom object - calls String.toString()
• null - null

Parameters:

target - the Kafka headers.

key - the key you want to add in the header.

value - the value you want to store in the header.

keys

public static Set<String> keys(org.apache.kafka.common.header.Headers headers)

Extract keys from Headers.

Parameters:

headers - the Kafka header.

Returns:

the keys.

generateMetadataKey

public static String generateMetadataKey(String key)

Generates key, used to identify Axon metadata in RecordHeader.

Parameters:

key - the actual key.

Returns:

the generated key.

extractKey

public static String extractKey(String metadataKey)

Extracts actual key name used to send Axon metadata. E.g."axon-metadata-foo" will extract foo

Parameters:

metadataKey - the generated metadata key.

Returns:

the actual key name.

extractAxonMetadata

public static Map<String,Object> extractAxonMetadata(org.apache.kafka.common.header.Headers target)

Extract axon metadata(if any) attached with Headers.

Parameters:

target - kafka header.

Returns:

the metadata.

toHeaders

public static org.apache.kafka.common.header.Headers toHeaders(EventMessage<?> eventMessage,
                                                               SerializedObject<byte[]> serializedObject,
                                                               BiFunction<String,Object,org.apache.kafka.common.header.internals.RecordHeader> headerValueMapper)

Generates Headers based on EventMessage and SerializedObject.

Parameters:

eventMessage - provides event data.

serializedObject - payload.

headerValueMapper - function for converting value to bytes. Since RecordHeader can handle only bytes this function needs to define the logic how to convert a given value to bytes. See byteMapper() for sample implementation.

Returns:

the Headers.

byteMapper

public static BiFunction<String,Object,org.apache.kafka.common.header.internals.RecordHeader> byteMapper()

Handles value conversion to bytes.

Returns:

the mapping function.