Testing RabbitMQ Client in Java

public class RabbitClient implements Supplier<String>, Consumer<String> {
    private static final Logger logger = LoggerFactory.getLogger(RabbitClient.class);
    private static final String EXCHANGE_TYPE = "direct";

    private static final boolean QUEUE_DURABLE = true;
    private static final boolean QUEUE_EXCLUSIVE = false;
    private static final boolean QUEUE_AUTODELETE = false;

    public static class Config {
        final String host;
        final int port;
        final String virtualHost;
        final String username;
        final String password;
        final String queueName;
        final String exchangeName;
        final String routingKey;
        final boolean useSsl;

        public Config(
            String host,
            int port,
            String virtualHost,
            String username,
            String password,
            String queueName,
            String exchangeName,
            String routingKey,
            boolean useSsl
        ) {
            this.host = host;
            this.port = port;
            this.virtualHost = virtualHost;
            this.username = username;
            this.password = password;
            this.queueName = queueName;
            this.exchangeName = exchangeName;
            this.routingKey = routingKey;
            this.useSsl = useSsl;
        }
    }

    private final RabbitClient.Config config;
    private final ConnectionFactory connectionFactory;

    public RabbitClient(final RabbitClient.Config config) {
        this.config = config;
        this.connectionFactory = new ConnectionFactory();
        this.connectionFactory.setUsername(config.username);
        this.connectionFactory.setPassword(config.password);
        this.connectionFactory.setVirtualHost(config.virtualHost);
        this.connectionFactory.setHost(config.host);
        this.connectionFactory.setPort(config.port);
        if (config.useSsl) {
            try {
                this.connectionFactory.useSslProtocol();
            } catch (Exception e) {
                logger.error("Error when enabling SSL", e);
            }
        }
    }

    private String withConnection(Function<Connection, String> block) {
        try (Connection connection = connectionFactory.newConnection()) {
            return block.apply(connection);
        } catch (Exception e) {
            logger.error("Error during using rabbit connection", e);
            return null;
        }
    }

    @Override
    public void accept(String payload) {
        withConnection(c -> {
            try(Channel channel = c.createChannel()) {
                channel.exchangeDeclare(config.exchangeName, EXCHANGE_TYPE, true);
                channel.queueDeclare(config.queueName, QUEUE_DURABLE, QUEUE_EXCLUSIVE, QUEUE_AUTODELETE, null);
                channel.queueBind(config.queueName, config.exchangeName, config.routingKey);
                channel.basicPublish(
                    config.exchangeName,
                    config.routingKey,
                    true,
                    MessageProperties.PERSISTENT_TEXT_PLAIN,
                    payload.getBytes());
            } catch (Exception e) {
                logger.error("Creating rabbit channel failed", e);
            }
            return null;
        });
    }

    @Override
    public String get() {
        return withConnection(c -> {
            try(Channel channel = c.createChannel()) {
                channel.exchangeDeclare(config.exchangeName, EXCHANGE_TYPE, true);
                channel.queueDeclare(config.queueName, QUEUE_DURABLE, QUEUE_EXCLUSIVE, QUEUE_AUTODELETE, null);
                channel.queueBind(config.queueName, config.exchangeName, config.routingKey);
                GetResponse response = channel.basicGet(config.queueName, false);
                if (response != null) {
                    byte[] body = response.getBody();
                    long tag = response.getEnvelope().getDeliveryTag();
                    channel.basicAck(tag, false);
                    return new String(body);
                }
                return null;
            } catch (Exception e) {
                logger.error("Creating rabbit channel failed", e);
                return null;
            }
        });
    }
}

public class RabbitClientTests {
    private static final Logger logger = LoggerFactory.getLogger(RabbitClientTests.class);

    private static final RabbitClient.Config config = new RabbitClient.Config(
        "localhost", 5672, "/", "guest", "guest", "q", "X", "key", true);

    private static final ConnectionFactory cf = new ConnectionFactory();
    static {
        cf.setUsername(config.username);
        cf.setPassword(config.password);
        cf.setHost(config.host);
        cf.setPort(config.port);
        cf.setVirtualHost(config.virtualHost);
        if (config.useSsl) {
            try {
                cf.useSslProtocol();
            } catch (NoSuchAlgorithmException | KeyManagementException e) {
                logger.error("Error when enabling SSL", e);
            }
        }
    }

    private static final Broker broker = new Broker();

    @BeforeClass
    public static void beforeClass() throws Exception {
        final BrokerOptions brokerOptions = new BrokerOptions();
        brokerOptions.setConfigurationStoreLocation("qpid/broker-config.json");
        broker.startup(brokerOptions);
    }

    @AfterClass
    public static void afterClass() {
        broker.shutdown();
    }

    private static Channel getChannel() throws Exception {
      Connection connection = cf.newConnection();
      Channel channel = connection.createChannel();
      channel.exchangeDeclare(config.exchangeName, "direct", true);
      channel.queueDeclare(config.queueName, false, false, false, null);
      channel.queueBind(config.queueName, config.exchangeName, config.routingKey);
      return channel;
    }

    private static void send(String message) throws Exception {
        Channel channel = getChannel();
        channel.basicPublish(config.exchangeName, config.routingKey, null, message.getBytes());
    }

    private static String recv() throws Exception {
        Channel channel = getChannel();
        GetResponse response = channel.basicGet(config.queueName, true);
        return new String(response.getBody());
    }

    @Test
    public void receive() throws Exception {
        String key = "{}";
        Supplier<String> supplier = new RabbitClient(config);
        send(key);
        assertEquals(key, supplier.get());
    }

    @Test
    public void send() throws Exception {
        String key = "{}";
        Consumer<String> consumer = new RabbitClient(config);
        consumer.accept(key);
        assertEquals(recv(), key);
    }
}