private void sendPublishRequest() {
        List<Long> outstandingRequests = new LinkedList<>();

        //If we are waiting on a response and haven't received one, just wait until we do. A keep alive will be sent out eventually
        if (outstandingRequests.size() <= 1) {
            RequestHeader requestHeader = conversation.createRequestHeader(this.revisedCycleTime * 10);

            //Make a copy of the outstanding requests, so it isn't modified while we are putting the ack list together.
            List<SubscriptionAcknowledgement> acks = new ArrayList<>(outstandingAcknowledgements);
            // do not send -1 when requesting publish, the -1 value indicates NULL value
            // which might result in corruption of subscription for some servers
            int ackLength = acks.size();
            outstandingAcknowledgements.removeAll(acks);

            PublishRequest publishRequest = new PublishRequest(requestHeader, acks);
            // we work in external thread - we need to coordinate access to conversation pipeline
            RequestTransaction transaction = tm.startRequest();
            transaction.submit(() -> {
                logger.trace("Sent publish request with {} acks", ackLength);
                //  Create Consumer for the response message, error and timeout to be sent to the Secure Channel
                conversation.submit(publishRequest, PublishResponse.class).thenAccept(responseMessage -> {
                    outstandingRequests.remove(responseMessage.getResponseHeader().getRequestHandle());

                    for (long availableSequenceNumber : responseMessage.getAvailableSequenceNumbers()) {
                        outstandingAcknowledgements.add(new SubscriptionAcknowledgement(this.subscriptionId, availableSequenceNumber));
                    }

                    NotificationMessage message = responseMessage.getNotificationMessage();
                    if (message.getNotificationData() != null) {
                        for (ExtensionObject notificationMessage : message.getNotificationData()) {
                            ExtensionObjectDefinition notification = notificationMessage.getBody();
                            if (notification instanceof DataChangeNotification) {
                                logger.trace("Found a Data Change Notification");
                                DataChangeNotification data = (DataChangeNotification) notification;
                                if (!data.getMonitoredItems().isEmpty()) {
                                    onMonitoredValue(data.getMonitoredItems());
                                }
                            } else if (notification instanceof EventNotificationList) {
                                logger.trace("Found a Event Notification");
                                EventNotificationList data = (EventNotificationList) notification;
                                if (!data.getEvents().isEmpty()) {
                                    onEventNotification(data.getEvents());
                                }
                            } else {
                                logger.warn("Unsupported Notification type {}", notification.getClass().getName());
                            }
                        }
                    }
                }).whenComplete((result, error) -> {
                    if (error != null) {
                        logger.warn("Publish request of subscription {} resulted in error reported by server", subscriptionId, error);
                        transaction.failRequest(error);
                    } else {
                        logger.trace("Completed publish request for subscription {}", subscriptionId);
                        transaction.endRequest();
                    }
                });
                outstandingRequests.add(requestHeader.getRequestHandle());
            });
        }
    }
Main subscriber loop. For subscription, we still need to send a request the server on every cycle.
Which includes a request for an update of the previously agreed upon list of tags.
The server will respond at most once every cycle.
plc4j/drivers/opcua/src/main/java/org/apache/plc4x/java/opcua/protocol/OpcuaSubscriptionHandle.java