<commit_before>package com.deutscheboerse.risk.dave.amqp;

import com.deutscheboerse.risk.dave.config.AmqpConfig;
import io.vertx.codegen.annotations.Nullable;
import io.vertx.core.AsyncResult;
import io.vertx.core.Future;
import io.vertx.core.Handler;
import io.vertx.core.Vertx;
import io.vertx.core.logging.Logger;
import io.vertx.core.logging.LoggerFactory;
import io.vertx.proton.*;

public class AmqpClient {
    private static final Logger LOG = LoggerFactory.getLogger(AmqpClient.class);

    private static final int FULL_FLOW_CREDIT = 2;
    private static final int THROTTLE_FLOW_CREDIT = 1;

    private final Vertx vertx;
    private final AmqpConfig config;
    private final String verticleName;
    private final String queueName;
    private final ProtonClient protonClient;

    private ProtonConnection protonBrokerConnection;
    private ProtonReceiver protonBrokerReceiver;
    private boolean draining;
    private Runnable queuedFlowRequest = null;

    private Handler<AsyncResult<Void>> connectHandler;
    private ProtonMessageHandler deliveryHandler;
    private Handler<Void> disconnectHandler;

    public AmqpClient(Vertx vertx, AmqpConfig config, String verticleName, String queueName) {
        this.vertx = vertx;
        this.config = config;
        this.protonClient = ProtonClient.create(vertx);
        this.verticleName = verticleName;
        this.queueName = queueName;
    }

    public void connect() {
        createBrokerConnection()
                .compose(i -> setupDisconnectHandler())
                .compose(i -> createAmqpReceiver())
                .setHandler(this.connectHandler);
    }

    public AmqpClient setConnectHandler(Handler<AsyncResult<Void>> handler) {
        this.connectHandler = handler;
        return this;
    }

    public AmqpClient setDeliveryHandler(ProtonMessageHandler handler) {
        this.deliveryHandler = handler;
        return this;
    }

    public AmqpClient setDisconnectHandler(Handler<Void> handler) {
        this.disconnectHandler = handler;
        return this;
    }

    public void stop() {
        setFlowRequest(() -> {
            draining = true;
            protonBrokerReceiver.drain(0, drainCompleteHandler());
        });
    }

    public void throttle() {
        setFlowRequest(() -> protonBrokerReceiver.flow(THROTTLE_FLOW_CREDIT));
    }

    public void run() {
        setFlowRequest(() -> protonBrokerReceiver.flow(FULL_FLOW_CREDIT));
    }

    public void increaseCreditBy(int amount) {
        setFlowRequest(() -> protonBrokerReceiver.flow(amount));
    }

    public void disconnect() {
        this.connectHandler = null;
        this.deliveryHandler = null;
        this.disconnectHandler = null;
        if (this.protonBrokerReceiver != null) {
            this.protonBrokerReceiver.close();
        }
        if (this.protonBrokerConnection != null) {
            this.protonBrokerConnection.disconnect();
        }
    }

    private Future<Void> createBrokerConnection() {
        Future<Void> createBrokerConnectionFuture = Future.future();

        Future<ProtonConnection> connectFuture = Future.future();
        protonClient.connect(getClientOptions(),
                this.config.getHostname(),
                this.config.getPort(),
                this.config.getUsername(),
                this.config.getPassword(),
                connectFuture);

        connectFuture.compose(connectResult -> {
            // When the connection is established (connectFuture), execute this
            Future<ProtonConnection> openFuture = Future.future();
            connectResult.setContainer("mdh/marketDataLoader-" + this.verticleName).openHandler(openFuture).open();
            return openFuture;
        }).compose(openResult -> {
            // When the connection is open (openFuture), execute this
            this.protonBrokerConnection = openResult;
            createBrokerConnectionFuture.complete();
        }, createBrokerConnectionFuture);

        return createBrokerConnectionFuture;
    }

    private ProtonClientOptions getClientOptions() {
        return new ProtonClientOptions()
                .setReconnectAttempts(this.config.getReconnectAttempts())
                .setReconnectInterval(this.config.getReconnectTimeout());
    }

    private Future<Void> setupDisconnectHandler() {
        this.protonBrokerConnection.disconnectHandler(connection -> {
            LOG.info("{} disconnected ", this.verticleName);
            if (this.protonBrokerReceiver != null) {
                this.protonBrokerReceiver.close();
            }
            if (disconnectHandler != null) {
                disconnectHandler.handle(null);<commit_msg>Add heartbeat - hardcoded to 60 seconds #81<commit_after>package com.deutscheboerse.risk.dave.amqp;

import com.deutscheboerse.risk.dave.config.AmqpConfig;
import io.vertx.codegen.annotations.Nullable;
import io.vertx.core.AsyncResult;
import io.vertx.core.Future;
import io.vertx.core.Handler;
import io.vertx.core.Vertx;
import io.vertx.core.logging.Logger;
import io.vertx.core.logging.LoggerFactory;
import io.vertx.proton.*;

public class AmqpClient {
    private static final Logger LOG = LoggerFactory.getLogger(AmqpClient.class);

    private static final int FULL_FLOW_CREDIT = 2;
    private static final int THROTTLE_FLOW_CREDIT = 1;

    private final Vertx vertx;
    private final AmqpConfig config;
    private final String verticleName;
    private final String queueName;
    private final ProtonClient protonClient;

    private ProtonConnection protonBrokerConnection;
    private ProtonReceiver protonBrokerReceiver;
    private boolean draining;
    private Runnable queuedFlowRequest = null;

    private Handler<AsyncResult<Void>> connectHandler;
    private ProtonMessageHandler deliveryHandler;
    private Handler<Void> disconnectHandler;

    public AmqpClient(Vertx vertx, AmqpConfig config, String verticleName, String queueName) {
        this.vertx = vertx;
        this.config = config;
        this.protonClient = ProtonClient.create(vertx);
        this.verticleName = verticleName;
        this.queueName = queueName;
    }

    public void connect() {
        createBrokerConnection()
                .compose(i -> setupDisconnectHandler())
                .compose(i -> createAmqpReceiver())
                .setHandler(this.connectHandler);
    }

    public AmqpClient setConnectHandler(Handler<AsyncResult<Void>> handler) {
        this.connectHandler = handler;
        return this;
    }

    public AmqpClient setDeliveryHandler(ProtonMessageHandler handler) {
        this.deliveryHandler = handler;
        return this;
    }

    public AmqpClient setDisconnectHandler(Handler<Void> handler) {
        this.disconnectHandler = handler;
        return this;
    }

    public void stop() {
        setFlowRequest(() -> {
            draining = true;
            protonBrokerReceiver.drain(0, drainCompleteHandler());
        });
    }

    public void throttle() {
        setFlowRequest(() -> protonBrokerReceiver.flow(THROTTLE_FLOW_CREDIT));
    }

    public void run() {
        setFlowRequest(() -> protonBrokerReceiver.flow(FULL_FLOW_CREDIT));
    }

    public void increaseCreditBy(int amount) {
        setFlowRequest(() -> protonBrokerReceiver.flow(amount));
    }

    public void disconnect() {
        this.connectHandler = null;
        this.deliveryHandler = null;
        this.disconnectHandler = null;
        if (this.protonBrokerReceiver != null) {
            this.protonBrokerReceiver.close();
        }
        if (this.protonBrokerConnection != null) {
            this.protonBrokerConnection.disconnect();
        }
    }

    private Future<Void> createBrokerConnection() {
        Future<Void> createBrokerConnectionFuture = Future.future();

        Future<ProtonConnection> connectFuture = Future.future();
        protonClient.connect(getClientOptions(),
                this.config.getHostname(),
                this.config.getPort(),
                this.config.getUsername(),
                this.config.getPassword(),
                connectFuture);

        connectFuture.compose(connectResult -> {
            // When the connection is established (connectFuture), execute this
            Future<ProtonConnection> openFuture = Future.future();
            connectResult.setContainer("mdh/marketDataLoader-" + this.verticleName).openHandler(openFuture).open();
            return openFuture;
        }).compose(openResult -> {
            // When the connection is open (openFuture), execute this
            this.protonBrokerConnection = openResult;
            createBrokerConnectionFuture.complete();
        }, createBrokerConnectionFuture);

        return createBrokerConnectionFuture;
    }

    private ProtonClientOptions getClientOptions() {
        return new ProtonClientOptions()
                .setHeartbeat(60000)
                .setReconnectAttempts(this.config.getReconnectAttempts())
                .setReconnectInterval(this.config.getReconnectTimeout());
    }

    private Future<Void> setupDisconnectHandler() {
        this.protonBrokerConnection.disconnectHandler(connection -> {
            LOG.info("{} disconnected ", this.verticleName);
            if (this.protonBrokerReceiver != null) {
                this.protonBrokerReceiver.close();
            }
            if (disconnectHandler != null) {
                disconnectHandler.handle(null);
{'fraction_non_alphanumeric': 0.09131308856205693, 'fraction_numerical': 0.0015871336366522061, 'mean_word_length': 2.7374456306840647, 'pattern_counts': {'":': 0, '<': 25, '<?xml version=': 0, '>': 43, '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': '3871532', 'n_tokens_mistral': 2582, 'n_tokens_neox': 2473, 'n_words': 496}