import io.micronaut.core.bind.annotation.Bindable;

import java.lang.annotation.*;

@Documented
@Retention(RetentionPolicy.RUNTIME)
@Target({ElementType.PARAMETER})
@Bindable // (1)
public @interface Correlation {
}

import io.micronaut.core.bind.annotation.Bindable

import java.lang.annotation.*

@Documented
@Retention(RetentionPolicy.RUNTIME)
@Target([ElementType.PARAMETER])
@Bindable // (1)
@interface Correlation {
}

import io.micronaut.core.bind.annotation.Bindable


@MustBeDocumented
@Retention(AnnotationRetention.RUNTIME)
@Target(AnnotationTarget.VALUE_PARAMETER)
@Bindable // (1)
annotation class Correlation

import io.micronaut.core.convert.ArgumentConversionContext;
import io.micronaut.core.convert.ConversionService;
import io.micronaut.core.type.Argument;
import io.micronaut.mqtt.bind.AnnotatedMqttBinder;
import io.micronaut.mqtt.v5.bind.MqttV5BindingContext;
import jakarta.inject.Singleton;

import java.util.Optional;

@Singleton // (1)
public class CorrelationAnnotationBinder implements AnnotatedMqttBinder<MqttV5BindingContext, Correlation> { // (2)

    private final ConversionService conversionService;

    public CorrelationAnnotationBinder(ConversionService conversionService) { // (3)
        this.conversionService = conversionService;
    }

    @Override
    public Class<Correlation> getAnnotationType() {
        return Correlation.class;
    }

    @Override
    public void bindTo(MqttV5BindingContext context, Object value, Argument<Object> argument) {
        context.getProperties().setCorrelationData((byte[]) value); // (4)
    }

    @Override
    public Optional<Object> bindFrom(MqttV5BindingContext context, ArgumentConversionContext<Object> conversionContext) {
        return conversionService.convert(context.getProperties().getCorrelationData(), conversionContext); // (5)
    }
}

import io.micronaut.core.type.Argument
import io.micronaut.mqtt.bind.AnnotatedMqttBinder
import io.micronaut.mqtt.v5.bind.MqttV5BindingContext
import io.micronaut.core.convert.ArgumentConversionContext
import io.micronaut.core.convert.ConversionService

import jakarta.inject.Singleton

@Singleton // (1)
class CorrelationAnnotationBinder implements AnnotatedMqttBinder<MqttV5BindingContext, Correlation> { // (2)

    private final ConversionService conversionService

    CorrelationAnnotationBinder(ConversionService conversionService) { // (3)
        this.conversionService = conversionService
    }

    @Override
    Class<Correlation> getAnnotationType() {
        Correlation
    }

    @Override
    void bindTo(MqttV5BindingContext context, Object value, Argument<Object> argument) {
        context.properties.correlationData = (byte[]) value // (4)
    }

    @Override
    Optional<Object> bindFrom(MqttV5BindingContext context, ArgumentConversionContext<Object> conversionContext) {
        conversionService.convert(context.properties.correlationData, conversionContext) // (5)
    }
}

import io.micronaut.core.convert.ArgumentConversionContext
import io.micronaut.core.convert.ConversionService
import io.micronaut.core.type.Argument
import io.micronaut.mqtt.bind.AnnotatedMqttBinder
import io.micronaut.mqtt.v5.bind.MqttV5BindingContext
import java.util.*

import jakarta.inject.Singleton

@Singleton // (1)
class CorrelationAnnotationBinder(private val conversionService: ConversionService)// (3)
    : AnnotatedMqttBinder<MqttV5BindingContext, Correlation> { // (2)

    override fun getAnnotationType(): Class<Correlation> {
        return Correlation::class.java
    }

    override fun bindTo(context: MqttV5BindingContext, value: Any, argument: Argument<Any>) {
       context.properties.correlationData = value as? ByteArray // (4)
    }

    override fun bindFrom(context: MqttV5BindingContext, conversionContext: ArgumentConversionContext<Any>): Optional<Any> {
        return conversionService.convert(context.properties.correlationData, conversionContext) // (5)
    }
}

import io.micronaut.mqtt.annotation.MqttSubscriber;
import io.micronaut.mqtt.annotation.Topic;

import java.util.*;

@MqttSubscriber
public class ProductListener {

    Set<String> messages = Collections.synchronizedSet(new HashSet<>());

    @Topic("product")
    public void receive(byte[] data, @Correlation byte[] correlation) { // (1)
        messages.add(new String(correlation));
    }
}

import io.micronaut.mqtt.annotation.MqttSubscriber
import io.micronaut.mqtt.annotation.Topic

@MqttSubscriber
class ProductListener {

    Set<String> messages = Collections.synchronizedSet(new HashSet<>())

    @Topic("product")
    void receive(@Correlation byte[] correlation) { // (1)
        messages.add(new String(correlation))
    }
}

import io.micronaut.mqtt.annotation.MqttSubscriber
import io.micronaut.mqtt.annotation.Topic
import java.nio.charset.StandardCharsets

import java.util.Collections
import java.util.HashSet

@MqttSubscriber
class ProductListener {

    val messages: MutableSet<String> = Collections.synchronizedSet(HashSet())

    @Topic("product")
    fun receive(@Correlation data: ByteArray) { // (1)
        messages.add(String(data, StandardCharsets.UTF_8))
    }
}

import io.micronaut.mqtt.annotation.Topic;
import io.micronaut.mqtt.annotation.v5.MqttPublisher;

@MqttPublisher
public interface ProductClient {

    @Topic("product")
    void send(@Correlation byte[] correlation);
}

import io.micronaut.mqtt.annotation.Topic
import io.micronaut.mqtt.annotation.v5.MqttPublisher

@MqttPublisher
interface ProductClient {

    @Topic("product")
    void send(@Correlation byte[] correlation)
}

import io.micronaut.mqtt.annotation.v5.MqttPublisher
import io.micronaut.mqtt.annotation.Topic

@MqttPublisher
interface ProductClient {

    @Topic("product")
    fun send(@Correlation data: ByteArray)
}

import io.micronaut.core.annotation.NonNull;
import io.micronaut.core.annotation.Nullable;

public class ProductInfo {

    private String size;
    private Long count;
    private Boolean sealed;

    public ProductInfo(@Nullable String size, // (1)
                       @NonNull Long count, // (2)
                       @NonNull Boolean sealed) { // (3)
        this.size = size;
        this.count = count;
        this.sealed = sealed;
    }

    public String getSize() {
        return size;
    }

    public Long getCount() {
        return count;
    }

    public Boolean getSealed() {
        return sealed;
    }
}

import io.micronaut.core.annotation.NonNull
import io.micronaut.core.annotation.Nullable

class ProductInfo {

    private String size
    private Long count
    private Boolean seal

    ProductInfo(@Nullable String size, // (1)
                @NonNull Long count, // (2)
                @NonNull Boolean seal) { // (3)
        this.size = size
        this.count = count
        this.seal = seal
    }

    String getSize() {
        size
    }

    Long getCount() {
        count
    }

    Boolean getSealed() {
        seal
    }
}

class ProductInfo(val size: String?, // (1)
                  val count: Long, // (2)
                  val sealed: Boolean)// (3)

import io.micronaut.core.convert.ArgumentConversionContext;
import io.micronaut.core.convert.ConversionService;
import io.micronaut.core.type.Argument;
import io.micronaut.mqtt.bind.TypedMqttBinder;
import io.micronaut.mqtt.v5.bind.MqttV5BindingContext;
import jakarta.inject.Singleton;
import org.eclipse.paho.mqttv5.common.packet.UserProperty;

import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.stream.Collectors;

@Singleton // (1)
public class ProductInfoTypeBinder implements TypedMqttBinder<MqttV5BindingContext, ProductInfo> { //(2)

    private final ConversionService conversionService;

    ProductInfoTypeBinder(ConversionService conversionService) { //(3)
        this.conversionService = conversionService;
    }

    @Override
    public Argument<ProductInfo> getArgumentType() {
        return Argument.of(ProductInfo.class);
    }

    @Override
    public void bindTo(MqttV5BindingContext context, ProductInfo value, Argument<ProductInfo> argument) {
        List<UserProperty> userPropertiesList = context.getProperties().getUserProperties();
        if (value.getSize() != null) {
            userPropertiesList.add(new UserProperty("productSize", value.getSize()));
        }
        userPropertiesList.add(new UserProperty("productCount", value.getCount().toString())); // (4)
        userPropertiesList.add(new UserProperty("productSealed", value.getSealed().toString()));
    }

    @Override
    public Optional<ProductInfo> bindFrom(MqttV5BindingContext context, ArgumentConversionContext<ProductInfo> conversionContext) {
        List<UserProperty> userPropertiesList = context.getProperties().getUserProperties();
        Map<String, String> userProperties = userPropertiesList.stream()
                .collect(Collectors.toMap(UserProperty::getKey, UserProperty::getValue));
        String size = userProperties.get("productSize");
        Optional<Long> count = Optional.ofNullable(userProperties.get("productCount"))
                .flatMap(value -> conversionService.convert(value, Long.class));
        Optional<Boolean> sealed = Optional.ofNullable(userProperties.get("productSealed"))
                .flatMap(value -> conversionService.convert(value, Boolean.class));

        if (count.isPresent() && sealed.isPresent()) {
            return Optional.of(new ProductInfo(size, count.get(), sealed.get())); // (5)
        } else {
            return Optional.empty();
        }
    }
}

import io.micronaut.mqtt.bind.TypedMqttBinder
import io.micronaut.mqtt.v5.bind.MqttV5BindingContext
import io.micronaut.core.convert.ArgumentConversionContext
import io.micronaut.core.convert.ConversionService
import io.micronaut.core.type.Argument
import jakarta.inject.Singleton
import org.eclipse.paho.mqttv5.common.packet.UserProperty

import java.util.stream.Collectors

@Singleton // (1)
class ProductInfoTypeBinder implements TypedMqttBinder<MqttV5BindingContext, ProductInfo> { //(2)

    private final ConversionService conversionService

    ProductInfoTypeBinder(ConversionService conversionService) { //(3)
        this.conversionService = conversionService
    }

    @Override
    Argument<ProductInfo> getArgumentType() {
        return Argument.of(ProductInfo)
    }

    @Override
    void bindTo(MqttV5BindingContext context, ProductInfo value, Argument<ProductInfo> argument) {
        List<UserProperty> userPropertiesList = context.properties.userProperties
        if (value.size != null) {
            userPropertiesList.add(new UserProperty("productSize", value.size))
        }
        userPropertiesList.add(new UserProperty("productCount", value.count.toString())) // (4)
        userPropertiesList.add(new UserProperty("productSealed", value.sealed.toString()))
    }

    @Override
    Optional<ProductInfo> bindFrom(MqttV5BindingContext context, ArgumentConversionContext<ProductInfo> conversionContext) {
        List<UserProperty> userPropertiesList = context.properties.userProperties
        Map<String, String> userProperties = userPropertiesList.stream()
                .collect(Collectors.toMap(UserProperty::getKey, UserProperty::getValue))
        String size = userProperties.get("productSize")
        Optional<Long> count = Optional.ofNullable(userProperties.get("productCount"))
                .flatMap(value -> conversionService.convert(value, Long.class))
        Optional<Boolean> sealed = Optional.ofNullable(userProperties.get("productSealed"))
                .flatMap(value -> conversionService.convert(value, Boolean.class))

        if (count.isPresent() && sealed.isPresent()) {
            return Optional.of(new ProductInfo(size, count.get(), sealed.get())) // (5)
        } else {
            return Optional.empty()
        }
    }
}

import io.micronaut.core.convert.ArgumentConversionContext
import io.micronaut.core.convert.ConversionService
import io.micronaut.core.type.Argument
import io.micronaut.mqtt.bind.*
import io.micronaut.mqtt.v5.bind.MqttV5BindingContext
import jakarta.inject.Singleton
import org.eclipse.paho.mqttv5.common.packet.UserProperty

import java.util.Optional
import java.util.stream.Collectors

@Singleton // (1)
class ProductInfoTypeBinder constructor(private val conversionService: ConversionService) //(3)
    : TypedMqttBinder<MqttV5BindingContext, ProductInfo> { // (2)

    override fun bindTo(context: MqttV5BindingContext, value: ProductInfo, argument: Argument<ProductInfo>) {
        val userPropertiesList = context.properties.userProperties
        if (value.size != null) {
            userPropertiesList.add(UserProperty("productSize", value.size)) // (4)
        }
        userPropertiesList.add(UserProperty("productCount", value.count.toString()))
        userPropertiesList.add(UserProperty("productSealed", value.sealed.toString()))
    }

    override fun bindFrom(context: MqttV5BindingContext, conversionContext: ArgumentConversionContext<ProductInfo>): Optional<ProductInfo> {
        val userPropertiesList = context.properties.userProperties
        val userProperties = userPropertiesList.stream()
                .collect(Collectors.toMap(UserProperty::getKey, UserProperty::getValue))
        val size = userProperties.get("productSize")
        val count = Optional.ofNullable(userProperties.get("productCount"))
                .flatMap { value -> conversionService.convert(value, Long::class.java) }
        val sealed = Optional.ofNullable(userProperties.get("productSealed"))
                .flatMap { value -> conversionService.convert(value, Boolean::class.java) }

        if (count.isPresent && sealed.isPresent) {
            return Optional.of(ProductInfo(size, count.get(), sealed.get())) // (5)
        } else {
            return Optional.empty()
        }
    }

    override fun getArgumentType(): Argument<ProductInfo> {
        return Argument.of(ProductInfo::class.java)
    }

}