<commit_before>package org.iotbricks.core.amqp.transport.proton;

import java.nio.ByteBuffer;
import java.util.HashMap;
import java.util.function.BiConsumer;
import java.util.function.Function;

import org.apache.qpid.proton.amqp.Binary;
import org.apache.qpid.proton.amqp.messaging.AmqpValue;
import org.apache.qpid.proton.amqp.messaging.ApplicationProperties;
import org.apache.qpid.proton.amqp.messaging.Data;
import org.apache.qpid.proton.amqp.messaging.Rejected;
import org.apache.qpid.proton.amqp.messaging.Section;
import org.apache.qpid.proton.amqp.transport.DeliveryState;
import org.apache.qpid.proton.message.Message;
import org.iotbricks.core.amqp.transport.AmqpTransport;
import org.iotbricks.core.amqp.transport.RequestInstance;
import org.iotbricks.core.amqp.transport.ResponseHandler;
import org.iotbricks.core.amqp.transport.proton.ProtonTransport.ProtonRequest;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import io.glutamate.util.concurrent.CloseableCompletionStage;
import io.vertx.core.Vertx;
import io.vertx.proton.ProtonDelivery;

public abstract class ProtonTransport<RI>
        extends AbstractAmqpTransport<ProtonRequest<RI>>
        implements AmqpTransport<Message> {

    private static final Logger logger = LoggerFactory.getLogger(ProtonTransport.class);

    protected static abstract class Builder<RI, C extends ProtonTransport<RI>, B extends Builder<RI, C, B>>
            extends AbstractAmqpTransport.Builder<ProtonRequest<RI>, ProtonTransport<?>, B> {

        protected Builder() {
        }

        protected Builder(final B other) {
            super(other);
        }
    }

    public abstract static class RequestBuilderImpl<R, RB extends RequestBuilderImpl<R, RB, RI>, RI>
            implements AmqpRequestBuilder<R, RB> {

        private final Function<RB, CloseableCompletionStage<R>> executor;

        protected final Message message;

        protected BiConsumer<Rejected, RequestInstance<?, ?>> rejectedConsumer;

        public RequestBuilderImpl(
                final Function<RB, CloseableCompletionStage<R>> executor) {
            this.executor = executor;
            this.message = Message.Factory.create();
        }

        @Override
        public CloseableCompletionStage<R> execute() {
            return this.executor.apply(builder());
        }

        protected Message getMessage() {
            return this.message;
        }

        public RB body(final Section body) {
            this.message.setBody(body);
            return builder();
        }

        public RB payload(final Section body) {
            this.message.setBody(body);
            return builder();
        }

        @Override
        public RB payload(final ByteBuffer payload) {
            this.message.setBody(new Data(Binary.create(payload)));
            return builder();
        }

        @Override
        public RB payload(final String payload) {
            this.message.setBody(new AmqpValue(payload));
            return builder();
        }

        @Override
        public RB subject(final String subject) {
            this.message.setSubject(subject);
            return builder();
        }

        @SuppressWarnings("unchecked")
        @Override
        public RB applicationProperty(final String key, final Object value) {
            ApplicationProperties properties = this.message.getApplicationProperties();
            if (properties == null) {
                properties = new ApplicationProperties(new HashMap<>());
            }
            properties.getValue().put(key, value);
            return builder();
        }

        @Override
        public RB rejected(final BiConsumer<Rejected, RequestInstance<?, ?>> consumer) {
            this.rejectedConsumer = consumer;
            return builder();
        }

        public RI buildInformation() {
            return null;
        }
    }

    protected ProtonTransport(final Vertx vertx, final Function<RI, String> addressProvider,
            final Builder<RI, ? extends ProtonTransport<?>, ?> options) {
        super(vertx, request -> addressProvider.apply(request.getInformation()), options);
    }
<commit_msg>Forgot to assign the newly created application properties, fixed.<commit_after>package org.iotbricks.core.amqp.transport.proton;

import java.nio.ByteBuffer;
import java.util.HashMap;
import java.util.function.BiConsumer;
import java.util.function.Function;

import org.apache.qpid.proton.amqp.Binary;
import org.apache.qpid.proton.amqp.messaging.AmqpValue;
import org.apache.qpid.proton.amqp.messaging.ApplicationProperties;
import org.apache.qpid.proton.amqp.messaging.Data;
import org.apache.qpid.proton.amqp.messaging.Rejected;
import org.apache.qpid.proton.amqp.messaging.Section;
import org.apache.qpid.proton.amqp.transport.DeliveryState;
import org.apache.qpid.proton.message.Message;
import org.iotbricks.core.amqp.transport.AmqpTransport;
import org.iotbricks.core.amqp.transport.RequestInstance;
import org.iotbricks.core.amqp.transport.ResponseHandler;
import org.iotbricks.core.amqp.transport.proton.ProtonTransport.ProtonRequest;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import io.glutamate.util.concurrent.CloseableCompletionStage;
import io.vertx.core.Vertx;
import io.vertx.proton.ProtonDelivery;

public abstract class ProtonTransport<RI>
        extends AbstractAmqpTransport<ProtonRequest<RI>>
        implements AmqpTransport<Message> {

    private static final Logger logger = LoggerFactory.getLogger(ProtonTransport.class);

    protected static abstract class Builder<RI, C extends ProtonTransport<RI>, B extends Builder<RI, C, B>>
            extends AbstractAmqpTransport.Builder<ProtonRequest<RI>, ProtonTransport<?>, B> {

        protected Builder() {
        }

        protected Builder(final B other) {
            super(other);
        }
    }

    public abstract static class RequestBuilderImpl<R, RB extends RequestBuilderImpl<R, RB, RI>, RI>
            implements AmqpRequestBuilder<R, RB> {

        private final Function<RB, CloseableCompletionStage<R>> executor;

        protected final Message message;

        protected BiConsumer<Rejected, RequestInstance<?, ?>> rejectedConsumer;

        public RequestBuilderImpl(
                final Function<RB, CloseableCompletionStage<R>> executor) {
            this.executor = executor;
            this.message = Message.Factory.create();
        }

        @Override
        public CloseableCompletionStage<R> execute() {
            return this.executor.apply(builder());
        }

        protected Message getMessage() {
            return this.message;
        }

        public RB body(final Section body) {
            this.message.setBody(body);
            return builder();
        }

        public RB payload(final Section body) {
            this.message.setBody(body);
            return builder();
        }

        @Override
        public RB payload(final ByteBuffer payload) {
            this.message.setBody(new Data(Binary.create(payload)));
            return builder();
        }

        @Override
        public RB payload(final String payload) {
            this.message.setBody(new AmqpValue(payload));
            return builder();
        }

        @Override
        public RB subject(final String subject) {
            this.message.setSubject(subject);
            return builder();
        }

        @SuppressWarnings("unchecked")
        @Override
        public RB applicationProperty(final String key, final Object value) {
            ApplicationProperties properties = this.message.getApplicationProperties();
            if (properties == null) {
                properties = new ApplicationProperties(new HashMap<>());
                this.message.setApplicationProperties(properties);
            }
            properties.getValue().put(key, value);
            return builder();
        }

        @Override
        public RB rejected(final BiConsumer<Rejected, RequestInstance<?, ?>> consumer) {
            this.rejectedConsumer = consumer;
            return builder();
        }

        public RI buildInformation() {
            return null;
        }
    }

    protected ProtonTransport(final Vertx vertx, final Function<RI, String> addressProvider,
            final Builder<RI, ? extends ProtonTransport<?>, ?> options) {
        super(vertx, request -> addressProvider.apply(request.getInformation()), options);
    }

{'fraction_non_alphanumeric': 0.09669783406320274, 'fraction_numerical': 0.0004734288081429755, 'mean_word_length': 2.7472283813747227, 'pattern_counts': {'":': 0, '<': 55, '<?xml version=': 0, '>': 57, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5222274', 'n_tokens_mistral': 2279, 'n_tokens_neox': 2200, 'n_words': 473}