@Override
    public CompletableFuture<PlcSubscriptionResponse> subscribe(PlcSubscriptionRequest subscriptionRequest) {
        if (!isConnected()) {
            CompletableFuture<PlcSubscriptionResponse> future = new CompletableFuture<>();
            future.completeExceptionally(new PlcRuntimeException("Disconnected"));
            return future;
        }
        if (!isFeatureSupported()) {
            CompletableFuture<PlcSubscriptionResponse> future = new CompletableFuture<>();
            future.completeExceptionally(new PlcRuntimeException("Not Supported"));
            return future;
        }

        CompletableFuture<PlcSubscriptionResponse> future = new CompletableFuture<>();
        HashMap<String, PlcSubscriptionResponse> valuesResponse = new HashMap<>();
        HashMap<String, CompletableFuture<S7Message>> futures = new HashMap<>();

        //Initialize multiple requests.
        CompletableFuture<PlcSubscriptionResponse> response = new CompletableFuture<>();
        subscriptionRequest.getTagNames().forEach(fieldName -> futures.put(fieldName, new CompletableFuture<>()));

        futures.put("DATA_", new CompletableFuture<>());

        final DefaultPlcSubscriptionRequest defaultRequest = (DefaultPlcSubscriptionRequest) subscriptionRequest;
//        final S7PlcSubscriptionRequest request = (S7PlcSubscriptionRequest) defaultRequest;
        
        int tpduId = getTpduId();

        //The main task that runs the subscriptions.
        clientExecutorService.submit(() -> {
            final DefaultPlcSubscriptionTag sf = (DefaultPlcSubscriptionTag) subscriptionRequest.getTags().get(0);
            final S7SubscriptionTag tag = (S7SubscriptionTag) sf.getTag();

            S7Message s7Message = null;
            switch (tag.getTagType()) {
                case EVENT_SUBSCRIPTION:
                    s7Message = encodeEventSubscriptionRequest(defaultRequest, tpduId);
                    break;
                case EVENT_UNSUBSCRIPTION:
                    //encodeEventUnSubscriptionRequest(request, parameterItems, payloadItems);
                    break;
                case ALARM_ACK:
                    //encodeAlarmAckRequest(request, parameterItems, payloadItems);
                    break;
                case ALARM_QUERY:
                    s7Message = encodeAlarmQueryRequest(defaultRequest, tpduId);
                    break;
                case CYCLIC_SUBSCRIPTION:
                    s7Message = encodeCycledS7ANYSubscriptionRequest(defaultRequest, tpduId);
                    break;
                case CYCLIC_DB_SUBSCRIPTION:
                    s7Message = encodeCycledDBREADSubscriptionRequest(defaultRequest, tpduId);
                    break;
                case CYCLIC_UNSUBSCRIPTION:
                    //encodeCycledUnSubscriptionRequest(request, parameterItems, payloadItems);
                    break;
                default:
            }
            if (s7Message == null) {
                throw new PlcInvalidTagException("Unsupported tag of type: " + tag.getTagType());
            }

            TPKTPacket tpktPacket = new TPKTPacket(new COTPPacketData(null, s7Message, true, (byte) tpduId));

            // Start a new request-transaction (Is ended in the response-handler)
            RequestTransactionManager.RequestTransaction transaction = tm.startRequest();
            transaction.submit(() -> conversationContext.sendRequest(tpktPacket)
                .onTimeout(new TransactionErrorCallback<>(future, transaction))
                .onError(new TransactionErrorCallback<>(future, transaction))
                .expectResponse(TPKTPacket.class, REQUEST_TIMEOUT)
                .unwrap(TPKTPacket::getPayload)
                .only(COTPPacketData.class)
                .unwrap(COTPPacket::getPayload)
                .check(p -> p.getTpduReference() == tpduId)
                .handle(p -> {
                    // Finish the request-transaction.
                    transaction.endRequest();

                    try {
                        //future.complete(decodeEventSubscriptionRequest(tagName, p, subscriptionRequest));
                        futures.get("DATA_").complete(p);
                    } catch (Exception e) {
                        logger.warn("Error sending 'write' message: '{}'", e.getMessage(), e);
                    }
                }));

            try {
                S7Message responseMessage = futures.get("DATA_").get();
                S7ParameterUserData parameter = (S7ParameterUserData) responseMessage.getParameter();
                S7ParameterUserDataItemCPUFunctions msgParameter = (S7ParameterUserDataItemCPUFunctions) parameter.getItems().get(0);

                valuesResponse.put(Integer.toString(msgParameter.getSequenceNumber()),
                    decodeEventSubscriptionResponse(Integer.toString(msgParameter.getSequenceNumber()), subscriptionRequest, futures.get("DATA_").get()));
            } catch (Exception ex) {
                logger.warn(ex.toString());
            }

            try {
                HashMap<String, PlcResponseItem<PlcSubscriptionHandle>> values = new HashMap<>();
                valuesResponse.forEach((s, p) -> {
                    if (p != null)
                        values.putAll(((DefaultPlcSubscriptionResponse) p).getValues());
                });
                response.complete(new DefaultPlcSubscriptionResponse(subscriptionRequest, values));
            } catch (Exception ex) {
                logger.warn(ex.getMessage());
            }
        });

        return response;
    }
Maps the S7ReadResponse of a PlcReadRequest to a PlcReadResponse
plc4j/drivers/s7/src/main/java/org/apache/plc4x/java/s7/readwrite/protocol/S7ProtocolLogic.java