HiveMQ Extension SDK Services

...

@Override
public void extensionStart(final ExtensionStartInput extensionStartInput, final ExtensionStartOutput extensionStartOutput) {

    final String clientId = "client-123";
    final ClientService clientService = Services.clientService();
    CompletableFuture<Boolean> connectedFuture = clientService.isClientConnected(clientId);

    connectedFuture.whenComplete(new BiConsumer<Boolean, Throwable>() {
        @Override
        public void accept(Boolean connected, Throwable throwable) {
            if(throwable == null) {
                System.out.println("Client with id {" + clientId + "} is connected: " + connected);
            } else {
                //please use more sophisticated logging
                throwable.printStackTrace();
            }
        }
    });
}

...

...
@Override
public void extensionStart(final ExtensionStartInput extensionStartInput, final ExtensionStartOutput extensionStartOutput) {

    final String clientId = "my-client-id";
    final ClientService clientService = Services.clientService();
    CompletableFuture<Optional<SessionInformation>> sessionFuture = clientService.getSession(clientId);

    sessionFuture.whenComplete(new BiConsumer<Optional<SessionInformation>, Throwable>() {
        @Override
        public void accept(Optional<SessionInformation> sessionInformationOptional, Throwable throwable) {
            if(throwable == null) {

                if(sessionInformationOptional.isPresent()){
                    SessionInformation information = sessionInformationOptional.get();
                    System.out.println("Session Found");
                    System.out.println("ID: " + information.getClienIdentifier());
                    System.out.println("Connected: " + information.isConnected());
                    System.out.println("Session Expiry Interval " + information.getSessionExpiryInterval());
                } else {
                    System.out.println("No session found for client id: " + clientId);
                }

            } else {
                //please use more sophisticated logging
                throwable.printStackTrace();
            }
        }
    });

}

...

...

@Override
public void extensionStart(final ExtensionStartInput extensionStartInput, final ExtensionStartOutput extensionStartOutput) {

    final String clientId = "client-123";
    final ClientService clientService = Services.clientService();
    CompletableFuture<Boolean> disconnectFuture = clientService.disconnectClient(clientId, true);

    disconnectFuture.whenComplete(new BiConsumer<Boolean, Throwable>() {
        @Override
        public void accept(Boolean disconnected, Throwable throwable) {
            if(throwable == null) {
                if(disconnected){
                    System.out.println("Client was successfully disconnected and no Will message was sent");
                } else {
                    System.out.println("Client not found");
                }
            } else {
                //please use more sophisticated logging
                throwable.printStackTrace();
            }
        }
    });

}

...

...

@Override
public void extensionStart(final ExtensionStartInput extensionStartInput, final ExtensionStartOutput extensionStartOutput) {

    final String clientId = "client-123";
    final ClientService clientService = Services.clientService();
    CompletableFuture<Boolean> invalidateSessionFuture = clientService.invalidateSession(clientId);

    invalidateSessionFuture.whenComplete(new BiConsumer<Boolean, Throwable>() {
        @Override
        public void accept(Boolean disconnected, Throwable throwable) {
            if(throwable == null) {
                if(disconnected){
                    System.out.println("Client was disconnected");
                    System.out.println("Will message was sent");
                    System.out.println("Client session was removed");
                } else {
                    System.out.println("Client was offline");
                    System.out.println("Client session was removed");
                }
            } else {
                if(throwable instanceof NoSuchClientIdException){
                    System.out.println("Client not found");
                }
                //please use more sophisticated logging
                throwable.printStackTrace();
            }
        }
    });

}

...

...

@Override
public void extensionStart(final ExtensionStartInput extensionStartInput, final ExtensionStartOutput extensionStartOutput) {

    final ClientService clientService = Services.clientService();

    // this is the default executor but used as executor argument for demonstration purposes
    final Executor executor = Services.extensionExecutorService();

    final Pattern pattern = Pattern.compile("client-[1-9]+");

    CompletableFuture<Void> iterationFuture = clientService.iterateAllClients(
            new IterationCallback<SessionInformation>() {
                @Override
                public void iterate(IterationContext context, SessionInformation sessionInformation) {
                    final String clientIdentifier = sessionInformation.getClientIdentifier();
                    if (pattern.matcher(clientIdentifier).matches()) {
                        System.out.println("Found client for pattern " + clientIdentifier);
                        // abort the iteration if you are not interested in the remaining information as this saves resources
                        context.abortIteration();
                    }
                }
            }, executor);

    iterationFuture.whenComplete((ignored, throwable) -> {
        if (throwable == null) {
            System.out.println("Iterated all clients");
        } else {
            throwable.printStackTrace(); // please use more sophisticated logging
        }
    });
}

...

...

@Override
public void extensionStart(final ExtensionStartInput extensionStartInput, final ExtensionStartOutput extensionStartOutput) {

    final ClientService clientService = Services.clientService();

    final AtomicInteger counter = new AtomicInteger();

    CompletableFuture<Void> iterationFuture = clientService.iterateAllClients(
            new IterationCallback<SessionInformation>() {
                @Override
                public void iterate(IterationContext context, SessionInformation sessionInformation) {
                    if (sessionInformation.isConnected()) {
                        counter.incrementAndGet();
                    }
                }
            });

    iterationFuture.whenComplete((ignored, throwable) -> {
        if (throwable == null) {
            System.out.println("Connected clients: " + counter.get());
        } else {
            throwable.printStackTrace(); // please use more sophisticated logging
        }
    });
}

...

...

@Override
public void extensionStart(final ExtensionStartInput extensionStartInput, final ExtensionStartOutput extensionStartOutput) {

    iterate(0);
}

public void iterate(final int attempts) {

    final AtomicInteger counter = new AtomicInteger();

    CompletableFuture<Void> iterationFuture = Services.clientService().iterateAllClients(
            new IterationCallback<SessionInformation>() {
                @Override
                public void iterate(IterationContext context, SessionInformation sessionInformation) {
                    if (sessionInformation.isConnected()) {
                        counter.incrementAndGet();
                    }
                }
            });

    iterationFuture.whenComplete((ignored, throwable) -> {
        if (throwable == null) {
            System.out.println("Connected clients: " + counter.get());

        // in case the cluster topology changes during iteration, an IterationFailedException is thrown
        } else if (throwable instanceof IterationFailedException) {
            // only retry 3 times
            if (attempts < 3) {
                final int newAttemptCount = attempts + 1;
                Services.extensionExecutorService().schedule(() ->
                        iterate(newAttemptCount), newAttemptCount * 10, TimeUnit.SECONDS); // schedule retry with delay in case topology change is not over, else we would get another IterationFailedException
            } else {
                System.out.println("Could not fully iterate all clients.");
            }
        } else {
            throwable.printStackTrace(); // please use more sophisticated logging
        }
    });
}

...

...

@Override
public void extensionStart(final ExtensionStartInput extensionStartInput, final ExtensionStartOutput extensionStartOutput) {

    final String topic = "topic";
    final String clientId = "test-client";

    TopicSubscriptionBuilder subscriptionBuilder = Builders.topicSubscription()
            .topicFilter(topic)
            .noLocal(false)
            .retainAsPublished(true)
            .qos(Qos.AT_MOST_ONCE)
            .subscriptionIdentifier(1);


    CompletableFuture<Void> addFuture = Services.subscriptionStore().addSubscription(clientId, subscriptionBuilder.build());

    addFuture.whenComplete(new BiConsumer<Void, Throwable>() {
        @Override
        public void accept(Void aVoid, Throwable throwable) {

            if(throwable != null){
                throwable.printStackTrace();
                return;
            }

            System.out.println("Successfully added subscription for topic: " + topic + " | client: " + clientId);
        }
    });
}

...

...

@Override
public void extensionStart(final ExtensionStartInput extensionStartInput, final ExtensionStartOutput extensionStartOutput) {

    final SubscriptionStore subscriptionStore = Services.subscriptionStore();

    final String clientID = "test-client";

    final Set<TopicSubscription> topicSet = new HashSet<>();

    topicSet.add(Builders.topicSubscription().topicFilter("$share/group/topic1").build());
    topicSet.add(Builders.topicSubscription().topicFilter("topic2").build());
    topicSet.add(Builders.topicSubscription().topicFilter("topic3").build());

    final CompletableFuture<Void> addFuture = subscriptionStore.addSubscriptions(clientID, topicSet);

    addFuture.whenComplete(new BiConsumer<Void, Throwable>() {
        @Override
        public void accept(final Void result, final Throwable throwable) {

            if(throwable != null){
                throwable.printStackTrace();
                return;
            }

            System.out.println("Successfully added subscriptions to client: " + clientID);

        }
    });
}

...

...

@Override
public void extensionStart(final ExtensionStartInput extensionStartInput, final ExtensionStartOutput extensionStartOutput) {

    final String topic = "topic";
    final String clientId = "test-client";

    CompletableFuture<Void> removeFuture = Services.subscriptionStore().removeSubscription(clientId, topic);

    removeFuture.whenComplete(new BiConsumer<Void, Throwable>() {
        @Override
        public void accept(Void aVoid, Throwable throwable) {

            if(throwable != null){
                throwable.printStackTrace();
                return;
            }

            System.out.println("Successfully removed subscription for topic: " + topic + " | client: " + clientId);
        }
    });
}

...

...

@Override
public void extensionStart(final ExtensionStartInput extensionStartInput, final ExtensionStartOutput extensionStartOutput) {

    final SubscriptionStore subscriptionStore = Services.subscriptionStore();

    final String clientID = "test-client";

    final Set<String> topicSet = new HashSet<>();
    topicSet.add("$share/group/topic1");
    topicSet.add("topic2");
    topicSet.add("topic3");

    final CompletableFuture<Void> removeFuture = subscriptionStore.removeSubscriptions(clientID, topicSet);

    removeFuture.whenComplete(new BiConsumer<Void, Throwable>() {
        @Override
        public void accept(final Void result, final Throwable throwable) {

            if(throwable != null){
                throwable.printStackTrace();
                return;
            }

            System.out.println("Successfully removed subscriptions for topics: " + topicSet + " | from client: " + clientID);

        }
    });
}

...

...

@Override
public void extensionStart(final ExtensionStartInput extensionStartInput, final ExtensionStartOutput extensionStartOutput) {

    final String clientId = "test-client";

    CompletableFuture<Set<TopicSubscription>> getFuture = Services.subscriptionStore().getSubscriptions(clientId);

    getFuture.whenComplete(new BiConsumer<Set<TopicSubscription>, Throwable>() {
        @Override
        public void accept(Set<TopicSubscription> topicSubscriptions, Throwable throwable) {

            if(throwable != null){
                throwable.printStackTrace();
                return;
            }

            if(topicSubscriptions.isEmpty()){
                System.out.println("Found no subscriptions for client: " + clientId);
                return;
            }

            System.out.println("Found subscriptions for client: " + clientId);

            for (TopicSubscription topicSubscription : topicSubscriptions) {
                System.out.println("---------------------");
                System.out.println("Topic :" + topicSubscription.getTopicFilter());
                System.out.println("Qos :" + topicSubscription.getQos().getQosNumber());
                System.out.println("No local :" + topicSubscription.getNoLocal());
                System.out.println("Retain as published :" + topicSubscription.getRetainAsPublished());
                System.out.println("Subscription identifier :" + topicSubscription.getSubscriptionIdentifier());
            }

        }
    });
}

...

...

@Override
public void extensionStart(final ExtensionStartInput extensionStartInput, final ExtensionStartOutput extensionStartOutput) {

    final SubscriptionStore subscriptionStore = Services.subscriptionStore();

    // this is the default executor but used as executor argument for demonstration purposes
    final Executor executor = Services.extensionExecutorService();

    final Pattern pattern = Pattern.compile("client-[1-9]+");

    CompletableFuture<Void> iterationFuture = subscriptionStore.iterateAllSubscriptions(
            new IterationCallback<SubscriptionsForClientResult>() {
                @Override
                public void iterate(IterationContext context, SubscriptionsForClientResult subscriptionsForClient) {
                    final String clientIdentifier = subscriptionsForClient.getClientId();
                    final Set<TopicSubscription> subscriptions = subscriptionsForClient.getSubscriptions();
                    if (pattern.matcher(clientIdentifier).matches()) {
                        System.out.println("Found client with subscriptions " + subscriptions);
                        // abort the iteration if you are not interested in the remaining information as this saves resources
                        context.abortIteration();
                    }
                }
            },
            executor);

    iterationFuture.whenComplete((ignored, throwable) -> {
        if (throwable == null) {
            System.out.println("Iterated all subscriptions");
        } else {
            throwable.printStackTrace(); // please use more sophisticated logging
        }
    });
}

...

...

@Override
public void extensionStart(final ExtensionStartInput extensionStartInput, final ExtensionStartOutput extensionStartOutput) {

    final SubscriptionStore subscriptionStore = Services.subscriptionStore();

    final AtomicInteger counter = new AtomicInteger();

    CompletableFuture<Void> iterationFuture = subscriptionStore.iterateAllSubscriptions(
            new IterationCallback<SubscriptionsForClientResult>() {
                @Override
                public void iterate(IterationContext context, SubscriptionsForClientResult subscriptionsForClient) {
                    counter.addAndGet(subscriptionsForClient.getSubscriptions().size());
                }
            });

    iterationFuture.whenComplete((ignored, throwable) -> {
        if (throwable == null) {
            System.out.println("Number of subscriptions: " + counter.get());
        } else {
            throwable.printStackTrace(); // please use more sophisticated logging
        }
    });
}

...

...

@Override
public void extensionStart(final ExtensionStartInput extensionStartInput, final ExtensionStartOutput extensionStartOutput) {

    iterate(0);
}

public void iterate(final int attempts) {

    final AtomicInteger counter = new AtomicInteger();

    CompletableFuture<Void> iterationFuture = Services.subscriptionStore().iterateAllSubscriptions(
            new IterationCallback<SubscriptionsForClientResult>() {
                @Override
                public void iterate(IterationContext context, SubscriptionsForClientResult subscriptionsForClient) {
                    counter.addAndGet(subscriptionsForClient.getSubscriptions().size());
                }
            });

    iterationFuture.whenComplete((ignored, throwable) -> {
        if (throwable == null) {
            System.out.println("Number of subscriptions: " + counter.get());

        // in case the cluster topology changes during iteration, an IterationFailedException is thrown
        } else if (throwable instanceof IterationFailedException) {
            // only retry 3 times
            if (attempts < 3) {
                final int newAttemptCount = attempts + 1;
                Services.extensionExecutorService().schedule(() ->
                        iterate(newAttemptCount), newAttemptCount * 10, TimeUnit.SECONDS); // schedule retry with delay in case topology change is not over, else we would get another IterationFailedException
            } else {
                System.out.println("Could not fully iterate all clients.");
            }
        } else {
            throwable.printStackTrace(); // please use more sophisticated logging
        }
    });
}

...

...

@Override
public void extensionStart(final ExtensionStartInput extensionStartInput, final ExtensionStartOutput extensionStartOutput) {

    final SubscriptionStore subscriptionStore = Services.subscriptionStore();

    // this is the default executor but used as executor argument for demonstration purposes
    final Executor executor = Services.extensionExecutorService();

    final AtomicInteger counter = new AtomicInteger();

    CompletableFuture<Void> iterationFuture = subscriptionStore.iterateAllSubscribersWithTopicFilter(
            "example/topic",
            SubscriptionType.INDIVIDUAL,
            new IterationCallback<SubscriberWithFilterResult>() {
                @Override
                public void iterate(IterationContext context, SubscriberWithFilterResult subscriberWithFilter) {
                    counter.incrementAndGet();
                }
            },
            executor);

    iterationFuture.whenComplete((ignored, throwable) -> {
        if (throwable == null) {
            System.out.println("Number of subscribers with specified topic filter: " + counter.get());
        } else {
            throwable.printStackTrace(); // please use more sophisticated logging
        }
    });
}

...

...

@Override
public void extensionStart(final ExtensionStartInput extensionStartInput, final ExtensionStartOutput extensionStartOutput) {

    final SubscriptionStore subscriptionStore = Services.subscriptionStore();

    // this is the default executor but used as executor argument for demonstration purposes
    final Executor executor = Services.extensionExecutorService();

    final AtomicInteger counter = new AtomicInteger();

    CompletableFuture<Void> iterationFuture = subscriptionStore.iterateAllSubscribersForTopic(
            "example/topic",
            SubscriptionType.INDIVIDUAL,
            new IterationCallback<SubscriberWithFilterResult>() {
                @Override
                public void iterate(IterationContext context, SubscriberWithFilterResult subscriberWithFilter) {
                    counter.incrementAndGet();
                }
            },
            executor);

    iterationFuture.whenComplete((ignored, throwable) -> {
        if (throwable == null) {
            System.out.println("Number of subscribers for specified topic: " + counter.get());
        } else {
            throwable.printStackTrace(); // please use more sophisticated logging
        }
    });
}

...

...

@Override
public void extensionStart(final ExtensionStartInput extensionStartInput, final ExtensionStartOutput extensionStartOutput) {

    // 1. build the retained message via the RetainedPublishBuilder
    final RetainedPublishBuilder retainedPublishBuilder = Builders.retainedPublish();

    final RetainedPublish retainedMessage = retainedPublishBuilder
        .topic("add/message")
        .payload(ByteBuffer.wrap("test".getBytes()))
        .userProperty("reason","message-update")
        .qos(Qos.AT_LEAST_ONCE)
        .build();

    // 2. add the retained message (if a retained message already exists for the topic, it will be overwritten)
    final CompletableFuture<Void> addFuture = Services.retainedMessageStore().addOrReplace(retainedMessage);

    addFuture.whenComplete(new BiConsumer<Void, Throwable>() {
        @Override
        public void accept(Void aVoid, Throwable throwable) {

            if(throwable != null){
                throwable.printStackTrace();
                return;
            }

            // 3. log when the message was successfully added/replaced
            System.out.println("Successfully added retained message for topic: " + topic);
        }
    });
}

...

...

@Override
public void extensionStart(final ExtensionStartInput extensionStartInput, final ExtensionStartOutput extensionStartOutput) {

    final String topic = "topic";

    // 1. remove the retained message from the given topic
    final CompletableFuture<Void> removeFuture = Services.retainedMessageStore().remove(topic);

    removeFuture.whenComplete(new BiConsumer<Void, Throwable>() {
        @Override
        public void accept(Void aVoid, Throwable throwable) {

            if(throwable != null){
                throwable.printStackTrace();
                return;
            }

            // 2. log when the message was successfully removed (also happens when no retained message for that topic)
            System.out.println("Successfully removed retained message for topic: " + topic);
        }
    });
}

...

...

@Override
public void extensionStart(final ExtensionStartInput extensionStartInput, final ExtensionStartOutput extensionStartOutput) {

    final String topic = "topic";

    // 1. request retained message for topic
    final CompletableFuture<Optional<RetainedPublish>> getFuture = Services.retainedMessageStore().getRetainedMessage(topic);

    getFuture.whenComplete(new BiConsumer<Optional<RetainedPublish>, Throwable>() {
        @Override
        public void accept(Optional<RetainedPublish> retainedPublishOptional, Throwable throwable) {

            if (throwable != null) {
                throwable.printStackTrace();
                return;
            }

            // 2. check if a retained message exists for that topic
            if (!retainedPublishOptional.isPresent()) {
                System.out.println("Found no retained message for topic: " + topic);
                return;
            }

            // 3. log some information about the retained message
            final RetainedPublish retainedPublish = retainedPublishOptional.get();

            System.out.println("Found retained message for topic: " + topic);
            System.out.println("---------------------");
            System.out.println("Topic :" + retainedPublish.getTopic());
            System.out.println("Qos :" + retainedPublish.getQos().getQosNumber());

        }
    });
}

...

...

@Override
public void extensionStart(final ExtensionStartInput extensionStartInput, final ExtensionStartOutput extensionStartOutput) {

    // 1. request to delete all retained messages from the HiveMQ cluster
    final CompletableFuture<Void> clearFuture = Services.retainedMessageStore().clear();

    clearFuture.whenComplete(new BiConsumer<Void, Throwable>() {
        @Override
        public void accept(Void aVoid, Throwable throwable) {

            if(throwable != null){
                throwable.printStackTrace();
                return;
            }

            // 2. log when all retained messages were removed
            System.out.println("Successfully removed all retained messages");
        }
    });
}

...

...

@Override
public void extensionStart(final ExtensionStartInput extensionStartInput, final ExtensionStartOutput extensionStartOutput) {

    final RetainedMessageStore retainedMessageStore = Services.retainedMessageStore();

    // this is the default executor but used as executor argument for demonstration purposes
    final Executor executor = Services.extensionExecutorService();

    final Pattern pattern = Pattern.compile("sensor/id-[1-9]+");

    CompletableFuture<Void> iterationFuture = retainedMessageStore.iterateAllRetainedMessages(
            new IterationCallback<RetainedPublish>() {
                @Override
                public void iterate(final @NotNull IterationContext context, final @NotNull RetainedPublish retainedPublish) {
                    final String retainedPublishTopic = retainedPublish.getTopic();
                    if (pattern.matcher(retainedPublishTopic).matches()) {
                        System.out.println("Found retained messages with topic matching pattern " + retainedPublishTopic);
                        // abort the iteration if you are not interested in the remaining retained messages as this saves resources
                        context.abortIteration();
                    }
                }
            }, executor);

    iterationFuture.whenComplete((ignored, throwable) -> {
        if (throwable == null) {
            System.out.println("Iteration over retained messages complete"); // this will also be called if iteration is aborted manually
        } else {
            throwable.printStackTrace(); // please use more sophisticated logging
        }
    });
}
...

...

@Override
public void extensionStart(final ExtensionStartInput extensionStartInput, final ExtensionStartOutput extensionStartOutput) {

    final RetainedMessageStore retainedMessageStore = Services.retainedMessageStore();

    final AtomicInteger counter = new AtomicInteger();

    CompletableFuture<Void> iterationFuture = retainedMessageStore.iterateAllRetainedMessages(
            new IterationCallback<RetainedPublish>() {
                @Override
                public void iterate(final @NotNull IterationContext context, final @NotNull RetainedPublish retainedPublish) {
                    if (retainedPublish.getQos() == Qos.EXACTLY_ONCE) {
                            counter.incrementAndGet();
                    }
                }
            });

    iterationFuture.whenComplete((ignored, throwable) -> {
        if (throwable == null) {
            System.out.println("Retained messages with QoS level 2: " + counter.get());
        } else {
            throwable.printStackTrace(); // please use more sophisticated logging
        }
    });
}

...

...

@Override
public void extensionStart(final ExtensionStartInput extensionStartInput, final ExtensionStartOutput extensionStartOutput) {

    iterate(0);
}

public void iterate(final int attempts) {

    final AtomicInteger counter = new AtomicInteger();

    CompletableFuture<Void> iterationFuture = Services.retainedMessageStore().iterateAllRetainedMessages(
            new IterationCallback<RetainedPublish>() {
                @Override
                public void iterate(final @NotNull IterationContext context, final @NotNull RetainedPublish retainedPublish) {
                    if (retainedPublish.getQos() == Qos.EXACTLY_ONCE) {
                        counter.incrementAndGet();
                    }
                }
            });

    iterationFuture.whenComplete((ignored, throwable) -> {
        if (throwable == null) {
            System.out.println("Retained messages with QoS level 2: " + counter.get());

            // in case the cluster topology changes during iteration, an IterationFailedException is thrown
        } else if (throwable instanceof IterationFailedException) {
            // only retry 3 times
            if (attempts < 3) {
                final int newAttemptCount = attempts + 1;
                Services.extensionExecutorService().schedule(() ->
                        iterate(newAttemptCount), newAttemptCount * 10, TimeUnit.SECONDS); // schedule retry with delay in case topology change is not over, else we would get another IterationFailedException
            } else {
                System.out.println("Could not fully iterate all retained messages.");
            }
        } else {
            throwable.printStackTrace(); // please use more sophisticated logging
        }
    });
}

...

Services.eventRegistry().setClientLifecycleEventListener(input1 -> new ClientLifecycleEventListener() {
@Override
public void onMqttConnectionStart(final @NotNull ConnectionStartInput connectionStartInput) {
    // access the ConnectionAttributeStore via the ConnectionInformation from the ConnectionStartInput
    final ConnectionAttributeStore connectionAttributeStore = connectionStartInput.getConnectionInformation().getConnectionAttributeStore();
    // this example uses an empty byte array for the binary data
    byte[] binaryData = new byte[1000];
    // wrap the byte array inside a ByteBuffer
    ByteBuffer byteBuffer = ByteBuffer.wrap(binaryData);
    // set the created ByteBuffer as the value for the key "my data"
    connectionAttributeStore.put("my data", byteBuffer);
}

Services.eventRegistry().setClientLifecycleEventListener(input1 -> new ClientLifecycleEventListener() {
@Override
public void onMqttConnectionStart(final @NotNull ConnectionStartInput connectionStartInput) {
    // access the Connection Attribute Store via the connection information from the ConnectionStartInput interface
    final ConnectionAttributeStore connectionAttributeStore = connectionStartInput.getConnectionInformation().getConnectionAttributeStore();
    // use the putAsString convenience method
    connectionAttributeStore.putAsString("my data", "my value");
}

Services.eventRegistry().setClientLifecycleEventListener(input1 -> new ClientLifecycleEventListener() {
@Override
public void onMqttConnectionStart(final @NotNull ConnectionStartInput connectionStartInput) {
    // access the Connection Attribute Store via the connection information from the ConnectionStartInput interface
    final ConnectionAttributeStore connectionAttributeStore = connectionStartInput.getConnectionInformation().getConnectionAttributeStore();
    // use a special charset in conjunction with the putAsString convenience method
    connectionAttributeStore.putAsString("my key", "my value", StandardCharsets.ISO_8859_1);
}

Services.eventRegistry().setClientLifecycleEventListener(input1 -> new ClientLifecycleEventListener() {
    @Override
    public void onMqttConnectionStart(final @NotNull ConnectionStartInput connectionStartInput) {
        // access the Connection Attribute Store via the connection information from the ConnectionStartInput interface
        final ConnectionAttributeStore connectionAttributeStore = connectionStartInput.getConnectionInformation().getConnectionAttributeStore();

        final Optional<ByteBuffer> bufferOptional = connectionAttributeStore.get("exampleKey");
        // verify that a value is set for the given key
        if(bufferOptional.isEmpty()){
            // If no value is present, return to handle the missing value. Another option is to set the value.
            return;
        }

        // Retrieve the ByteBuffer from the optional. This operation is safe due to the previous verification that the value is present.
        final ByteBuffer byteBuffer = bufferOptional.get();

        // CAUTION: Because the ByteBuffer is read-only, you must copy the buffer to a new byte array:
        final ByteBuffer rewind = byteBuffer.asReadOnlyBuffer().rewind();
        final byte[] array = new byte[rewind.remaining()];
        rewind.get(array);

        // use the array
        System.out.println(new String(array, StandardCharsets.UTF_8));
    }
});

Services.eventRegistry().setClientLifecycleEventListener(input1 -> new ClientLifecycleEventListener() {
    @Override
    public void onMqttConnectionStart(final @NotNull ConnectionStartInput connectionStartInput) {
        // access the Connection Attribute Store via the connection information from the ConnectionStartInput interace
        final ConnectionAttributeStore connectionAttributeStore = connectionStartInput.getConnectionInformation().getConnectionAttributeStore();

        final Optional<Map<String, ByteBuffer>> optionalConnectionAttributes = connectionAttributeStore.getAll();

        // verify that connection attributes are present:
        if(optionalConnectionAttributes.isEmpty()){
            // If no value is present, return to handle the missing value. Another option is to set the value.
            return;
        }

        // this operation is safe due to the previous verification that the value is present
        final Map<String, ByteBuffer> allConnectionAttributes = optionalConnectionAttributes.get();

        // iterate the entries for the given client
        for (Map.Entry<String, ByteBuffer> entry : allConnectionAttributes.entrySet()) {
            // CAUTION: Because the ByteBuffer is read-only, you must copy the buffer to a new byte array:
            final ByteBuffer rewind = entry.getValue().asReadOnlyBuffer().rewind();
            final byte[] array = new byte[rewind.remaining()];
            rewind.get(array);
            System.out.println(entry.getKey() + ":" + new String(array));
        }
    }
});

Services.eventRegistry().setClientLifecycleEventListener(input1 -> new ClientLifecycleEventListener() {
    @Override
    public void onMqttConnectionStart(final @NotNull ConnectionStartInput connectionStartInput) {
        // access the Connection Attribute Store via the connection information from the ConnectionStartInput interface
        final ConnectionAttributeStore connectionAttributeStore = connectionStartInput.getConnectionInformation().getConnectionAttributeStore();

        final Optional<String> optionalValue = connectionAttributeStore.getAsString("exampleKey");
        // verify that a value is set for the given key
        if (optionalValue.isEmpty()) {
            // If no value is present, return to handle the missing value. Another option is to set the value.
            return;
        }

        // Retrieve the ByteBuffer from the optional. This operation is safe due to the previous verification that the value is present
        final String value = optionalValue.get();

        // use the value, for example, to print
        System.out.println(value);
    }
});

Services.eventRegistry().setClientLifecycleEventListener(input1 -> new ClientLifecycleEventListener() {
    @Override
    public void onMqttConnectionStart(final @NotNull ConnectionStartInput connectionStartInput) {
        // access the Connection Attribute Store via the connection information from the ConnectionStartInput interface
        final ConnectionAttributeStore connectionAttributeStore = connectionStartInput.getConnectionInformation().getConnectionAttributeStore();

        final Optional<String> optionalValue = connectionAttributeStore.getAsString("exampleKey", StandardCharsets.ISO_8859_1);
        // verify that a value is set for the given key
        if (optionalValue.isEmpty()) {
            // If no value is present, return to handle the missing value. Another option is to set the value.
            return;
        }

        // Retrieve the ByteBuffer from the optional. This operation is safe due to the previous verification that the value is present
        final String value = optionalValue.get();

        // use the value, for example, to print
        System.out.println(value);
    }
});

Services.eventRegistry().setClientLifecycleEventListener(input1 -> new ClientLifecycleEventListener() {
    @Override
    public void onMqttConnectionStart(final @NotNull ConnectionStartInput connectionStartInput) {
        // access the Connection Attribute Store via the connection information from the ConnectionStartInput interface
        final ConnectionAttributeStore connectionAttributeStore = connectionStartInput.getConnectionInformation().getConnectionAttributeStore();

        // Remove the value that is associated with the key. If a value is present, the value is returned within a optional.
        final Optional<ByteBuffer> optionalPreviousValue = connectionAttributeStore.remove("exampleKey");
        // test whether there is a previous value and print the value for reference
        if (optionalPreviousValue.isPresent()) {
            final ByteBuffer byteBuffer = optionalPreviousValue.get();
            // CAUTION: Because the ByteBuffer is read-only, you must copy the buffer to a new byte array:
            final ByteBuffer rewind = byteBuffer.asReadOnlyBuffer().rewind();
            final byte[] array = new byte[rewind.remaining()];
            rewind.get(array);
            System.out.println("Previous value:" + new String(array));
        }
    }
});

Services.eventRegistry().setClientLifecycleEventListener(input1 -> new ClientLifecycleEventListener() {
    @Override
    public void onMqttConnectionStart(final @NotNull ConnectionStartInput connectionStartInput) {
        // access the Connection Attribute Store via the connection information from the ConnectionStartInput interface
        final ConnectionAttributeStore connectionAttributeStore = connectionStartInput.getConnectionInformation().getConnectionAttributeStore();

        // remove all attributes
        connectionAttributeStore.clear();
    }
});

...
@Override
public void extensionStart(final ExtensionStartInput extensionStartInput, final ExtensionStartOutput extensionStartOutput) {

    // Create a publish builder
    final PublishBuilder publishBuilder = Builders.publish();
    final ByteBuffer payload = ByteBuffer.wrap("message".getBytes());
    // Build the publish
    publishBuilder.topic("topic").qos(Qos.AT_LEAST_ONCE).payload(payload);
    // Access the Publish Service
    final PublishService publishService = Services.publishService();
    // Asynchronously sent PUBLISH
    final CompletableFuture<Void> future = publishService.publish(publishBuilder.build());

    future.whenComplete((aVoid, throwable) -> {
        if(throwable == null) {
            System.out.println("Publish sent successfully");
        } else {
            //please use more sophisticated logging
            throwable.printStackTrace();
        }
    });
}
...

...
@Override
public void extensionStart(final ExtensionStartInput extensionStartInput, final ExtensionStartOutput extensionStartOutput) {

    // Create a publish builder
    final PublishBuilder publishBuilder = Builders.publish();
    final ByteBuffer payload = ByteBuffer.wrap("message".getBytes());
    // Build the publish
    publishBuilder.topic("topic").qos(Qos.AT_LEAST_ONCE).payload(payload);
    // Access the Publish Service
    final PublishService publishService = Services.publishService();
    // Asynchronously sent PUBLISH
    final CompletableFuture<Void> future = publishService.publish(publishBuilder.build());
    final String clientId = "client";
    final CompletableFuture<PublishToClientResult> future = publishService.publishToClient(publishBuilder.build(), clientId);

    future.whenComplete((result, throwable) -> {

        if (throwable == null) {
            if (result == PublishToClientResult.NOT_SUBSCRIBED) {
                System.out.println("Publish was not sent to client ("+clientId+
                        ") because it is not subscribed to a matching topic");
            } else {
                System.out.println("Publish sent successfully");
            }
        } else {
            //please use more sophisticated logging
            throwable.printStackTrace();
        }
    });
}

...

final ConsumerOptions consumerOptions = EnterpriseBuilders.consumerOptions().topicFilters("+/STAT").build();
final String consumerId = "myConsumerId";
final MessageConsumer myConsumer = new MySimpleConsumer();

final CompletableFuture<Void> registerFuture =
       EnterpriseServices.consumerService()
               .registerConsumer(consumerId, consumerOptions, new MessageConsumerProvider() {
                   @Override
                   public @NotNull MessageConsumer get() {
                       return myConsumer;
                   }
               });

registerFuture.whenComplete((aVoid, throwable) -> {
   if (throwable == null) {
       log.info("Consumer {} started.", consumerId);
   } else {
       log.warn("Not able to start  Consumer \"{}\", reason:", throwable.getMessage());
   }
});

public class MySimpleMessagingService {
   private static final Logger log = LoggerFactory.getLogger(MySimpleMessagingService.class);
   private static final @NotNull String MY_MESSAGE_ID = "MySimpleMessagingService";
   private final @NotNull ExtensionMessagingService extensionMessagingService;
   private final @NotNull ScheduledExecutorService scheduledExecutorService;

   public MySimpleMessagingService(final @NotNull ExtensionMessagingService extensionMessagingService,
                                   final @NotNull ScheduledExecutorService scheduledExecutorService) {
       this.extensionMessagingService = extensionMessagingService;
       this.scheduledExecutorService = scheduledExecutorService;
   }

   public void start() {
       extensionMessagingService.register(MY_MESSAGE_ID, new MySimpleMessagingService.MySimpleRespondCallback());
       scheduledExecutorService.scheduleAtFixedRate(this::send, 1, 1, TimeUnit.MINUTES);
   }

   public void stop() {
       extensionMessagingService.unregister(MY_MESSAGE_ID);
   }

   private void send() {
       byte[] data =  createDataToSend();
       //Send Message in the cluster and retrieve responses from other nodes
       final List<CompletableFuture<ExtensionMessageResponse>> responseList =
               extensionMessagingService.send(
                       MY_MESSAGE_ID,
                       data,
                       EnterpriseBuilders.extensionMessageOptions().mode(BroadcastMode.OTHERS).build());

       CompletableFuture.allOf(responseList.toArray(new CompletableFuture[]{}))
               .exceptionally(throwable -> null)
               .thenAccept(aVoid -> {
                   for (CompletableFuture<ExtensionMessageResponse> responseCompletableFuture : responseList) {
                       try {
                           final ExtensionMessageResponse response = responseCompletableFuture.get();
                           processDataFromResponse(response.getClusterNodeId(), response.getMessage());
                       } catch (ExecutionException | InterruptedException any) {
                           log.error(" Requesting response of data failed: ", any);
                       }
                   }
               });
   }

   private byte[] createDataToSend() {
       log.info("Create data to Send ");
       return RandomUtils.nextBytes(200);
   }

   private void processDataFromResponse(String clusterNodeId, byte[] message) {
       log.info("got response from {} ", clusterNodeId);
   }

   /**
    * The callback that receives an ExtensionMessage must respond in any case.
    */
   static class MySimpleRespondCallback implements ExtensionMessageCallback {
       boolean success = false;

       @Override
       public void receive(@NotNull ExtensionMessage message) {
           try {
               success = createRespond(message.receive());
           } finally {
               message.respond(new byte[]{(byte) (success ? 1 : 0)});
           }
       }

       private boolean createRespond(byte[] data) {
           return true;
       }
   }
}