refactor: Let startWriting() return CompletableFuture<Boolean> instead of boolean

Author: caoccao

ext/hivemq-edge-openapi-2025.19-SNAPSHOT.yaml

@@ -14,7 +14,7 @@ info:
     ## OpenAPI
     HiveMQ's REST API provides an OpenAPI 3.0 schema definition that can imported into popular API tooling (e.g. Postman) or can be used to generate client-code for multiple programming languages.
   title: HiveMQ Edge REST API
-  version: 2025.19-SNAPSHOT
+  version: 2025.14-SNAPSHOT
   x-logo:
     url: https://www.hivemq.com/img/svg/hivemq-bee.svg
 tags:
@@ -94,6 +94,9 @@ tags:
       These endpoints can be used to retrieve states of clients for the Data Hub.
 
     name: Data Hub - State
+  - description: |
+      These endpoints can be used to retrieve the different elements of the Edge topology
+    name: Domain
   - description: |
       These endpoints can be used to manage Pulse and its assets.
     name: Pulse
@@ -3697,7 +3700,7 @@ paths:
       summary: Add a new Adapter
       tags:
         - Protocol Adapters
-  /api/v1/management/protocol-adapters/northboundMappings:
+  /api/v1/management/protocol-adapters/mappings/northboundMappings:
     get:
       description: Get all northbound mappings
       operationId: get-northboundMappings
@@ -3706,12 +3709,13 @@ paths:
           content:
             application/json:
               schema:
-                $ref: '#/components/schemas/NorthboundMappingList'
+                $ref: '#/components/schemas/NorthboundMappingOwnerList'
           description: Success
-      summary: Get the mappings for northbound messages.
+      summary: Get all the northbound mappings.
       tags:
         - Protocol Adapters
-  /api/v1/management/protocol-adapters/southboundMappings:
+        - Domain
+  /api/v1/management/protocol-adapters/mappings/southboundMappings:
     get:
       description: Get all southbound mappings.
       operationId: get-southboundMappings
@@ -3720,11 +3724,12 @@ paths:
           content:
             application/json:
               schema:
-                $ref: '#/components/schemas/SouthboundMappingList'
+                $ref: '#/components/schemas/SouthboundMappingOwnerList'
           description: Success
-      summary: Get all  southbound mappings.
+      summary: Get all the southbound mappings.
       tags:
         - Protocol Adapters
+        - Domain
   /api/v1/management/protocol-adapters/status:
     get:
       description: Obtain the details.
@@ -3796,11 +3801,12 @@ paths:
                           node: ns=2;i=test2
                         name: tag2
               schema:
-                $ref: '#/components/schemas/DomainTagList'
+                $ref: '#/components/schemas/DomainTagOwnerList'
           description: Success
-      summary: Get the list of all domain tags created in this Edge instance
+      summary: Get the list of all tags created in this Edge instance
       tags:
         - Protocol Adapters
+        - Domain
   /api/v1/management/protocol-adapters/tags/{tagName}:
     get:
       description: Get a domain tag created in this Edge instance
@@ -4060,6 +4066,7 @@ paths:
       summary: Get the list of all topic filters created in this Edge instance
       tags:
         - Topic Filters
+        - Domain
     post:
       description: Add a new topic filter.
       operationId: add-topicFilters
@@ -4583,6 +4590,7 @@ paths:
       operationId: get-managed-assets
       tags:
         - Pulse
+        - Domain
       responses:
         '200':
           content:
@@ -6450,6 +6458,44 @@ components:
             $ref: '#/components/schemas/DomainTag'
       required:
         - items
+    NorthboundMappingOwnerList:
+      type: object
+      properties:
+        items:
+          type: array
+          description: List of result items that are returned by this endpoint
+          items:
+            type: object
+            required:
+              - adapterId
+              - mapping
+            properties:
+              adapterId:
+                type: string
+                description: The id of the adapter owning the mapping
+              mapping:
+                $ref: '#/components/schemas/NorthboundMapping'
+      required:
+        - items
+    SouthboundMappingOwnerList:
+      type: object
+      properties:
+        items:
+          type: array
+          description: List of result items that are returned by this endpoint
+          items:
+            type: object
+            required:
+              - adapterId
+              - mapping
+            properties:
+              adapterId:
+                type: string
+                description: The id of the adapter owning the mapping
+              mapping:
+                $ref: '#/components/schemas/SouthboundMapping'
+      required:
+        - items
     TagSchema:
       type: object
       properties:
@@ -6458,6 +6504,25 @@ components:
         protocolId:
           type: string
           description: The id assigned to the protocol adapter type
+    DomainTagOwnerList:
+      type: object
+      properties:
+        items:
+          type: array
+          description: List of result items that are returned by this endpoint
+          items:
+            type: object
+            required:
+              - adapterId
+              - mapping
+            properties:
+              adapterId:
+                type: string
+                description: The id of the adapter owning the tag
+              mapping:
+                $ref: '#/components/schemas/DomainTag'
+      required:
+        - items
     ProtocolAdapterCategory:
       type: object
       description: The category of the adapter

hivemq-edge/src/main/java/com/hivemq/bootstrap/factories/WritingServiceProvider.java

@@ -30,6 +30,7 @@
 import org.slf4j.LoggerFactory;
 
 import java.util.List;
+import java.util.concurrent.CompletableFuture;
 
 @Singleton
 public class WritingServiceProvider {
@@ -78,12 +79,13 @@ public boolean writingEnabled() {
 
 
         @Override
-        public boolean startWriting(
+        public CompletableFuture<Boolean> startWriting(
                 final @NotNull WritingProtocolAdapter writingProtocolAdapter,
                 final @NotNull ProtocolAdapterMetricsService protocolAdapterMetricsService,
                 final @NotNull List<InternalWritingContext> southboundMappings) {
             log.warn("No bidirectional module is currently installed. Writing to PLCs is currently not supported.");
-            return true;        }
+            return CompletableFuture.completedFuture(true);
+        }
 
         @Override
         public void stopWriting(

hivemq-edge/src/main/java/com/hivemq/protocols/InternalProtocolAdapterWritingService.java

@@ -19,15 +19,17 @@
 import com.hivemq.adapter.sdk.api.services.ProtocolAdapterWritingService;
 import com.hivemq.adapter.sdk.api.writing.WritingContext;
 import com.hivemq.adapter.sdk.api.writing.WritingProtocolAdapter;
+import com.hivemq.persistence.SingleWriterService;
 import org.jetbrains.annotations.NotNull;
 
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Future;
 
 public interface InternalProtocolAdapterWritingService extends ProtocolAdapterWritingService {
 
 
-    boolean startWriting(
+    @NotNull CompletableFuture<Boolean> startWriting(
             @NotNull WritingProtocolAdapter writingProtocolAdapter,
             @NotNull ProtocolAdapterMetricsService protocolAdapterMetricsService,
             @NotNull List<InternalWritingContext> writingContexts);

hivemq-edge/src/main/java/com/hivemq/protocols/ProtocolAdapterWrapper.java

@@ -48,6 +48,7 @@
 import java.util.List;
 import java.util.Optional;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutionException;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Function;
@@ -200,21 +201,25 @@ private void stopAfterFailedStart() {
         }
     }
 
-    private Optional<Throwable> attemptStartingConsumers(final boolean writingEnabled, final @NotNull EventService eventService) {
+    private @NotNull Optional<Throwable> attemptStartingConsumers(final boolean writingEnabled, final @NotNull EventService eventService) {
         try {
             //Adapter started successfully, now start the consumers
             createAndSubscribeTagConsumer();
             startPolling(protocolAdapterPollingService, eventService);
 
-            if(writingEnabled && isWriting()) {
+            if (writingEnabled && isWriting()) {
                 final var started = new AtomicBoolean(false);
                 protocolAdapterState.setConnectionStatusListener(status -> {
-                    if(status == ProtocolAdapterState.ConnectionStatus.CONNECTED) {
-                        if(started.compareAndSet(false, true)) {
-                            if(startWriting(protocolAdapterWritingService)) {
-                                log.info("Successfully started adapter with id {}", adapter.getId());
-                            } else {
-                                log.error("Protocol adapter start failed as data hub is not available.");
+                    if (status == ProtocolAdapterState.ConnectionStatus.CONNECTED) {
+                        if (started.compareAndSet(false, true)) {
+                            try {
+                                if (startWriting(protocolAdapterWritingService).get()) {
+                                    log.info("Successfully started adapter with id {}", adapter.getId());
+                                } else {
+                                    log.error("Protocol adapter start failed as data hub is not available.");
+                                }
+                            } catch (final Exception e) {
+                                log.error("Failed to start writing for adapter with id {}.", adapter.getId(), e);
                             }
                         }
                     }
@@ -443,7 +448,7 @@ private void stopPolling(
         }
     }
 
-    private @NotNull boolean startWriting(final @NotNull InternalProtocolAdapterWritingService protocolAdapterWritingService) {
+    private @NotNull CompletableFuture<Boolean> startWriting(final @NotNull InternalProtocolAdapterWritingService protocolAdapterWritingService) {
         log.debug("Start writing for protocol adapter with id '{}'", getId());
 
         final var southboundMappings = getSouthboundMappings();

hivemq-edge/src/main/java/com/hivemq/protocols/fsm/ProtocolAdapterWrapper.java renamed to hivemq-edge/src/main/java/com/hivemq/protocols/fsm/ProtocolAdapterInstance.java

@@ -18,13 +18,13 @@
 
 import org.jetbrains.annotations.NotNull;
 
-public class ProtocolAdapterWrapper {
+public class ProtocolAdapterInstance {
     protected @NotNull ProtocolAdapterState state;
     protected @NotNull ProtocolAdapterConnectionState connectionState;
 
-    public ProtocolAdapterWrapper() {
-        this.state = ProtocolAdapterState.Stopped;
-        this.connectionState = ProtocolAdapterConnectionState.Closed;
+    public ProtocolAdapterInstance() {
+        connectionState = ProtocolAdapterConnectionState.Closed;
+        state = ProtocolAdapterState.Stopped;
     }
 
     public @NotNull ProtocolAdapterState getState() {
@@ -35,15 +35,18 @@ public ProtocolAdapterWrapper() {
         return connectionState;
     }
 
-    public @NotNull ProtocolAdapterTransitionStatus transitionTo(final @NotNull ProtocolAdapterState protocolAdapterState) {
-        final ProtocolAdapterTransitionResult result = state.transition(protocolAdapterState, this);
-        state = result.state();
-        // Handle error (logging, throwing exception, etc.)
-        switch (result.status()) {
-            default -> {
-                // TODO
+    public synchronized @NotNull ProtocolAdapterTransitionResponse transitionTo(final @NotNull ProtocolAdapterState newState) {
+        final ProtocolAdapterTransitionResponse response = state.transition(newState, this);
+        if (response.status() == ProtocolAdapterTransitionStatus.Success) {
+            this.state = response.state();
+        } else {
+            // Handle error (logging, throwing exception, etc.)
+            switch (response.status()) {
+                default -> {
+                    // TODO
+                }
             }
         }
-        return result.status();
+        return response;
     }
 }

hivemq-edge/src/main/java/com/hivemq/protocols/fsm/ProtocolAdapterOperator.java

@@ -0,0 +1,20 @@
+/*
+ *  Copyright 2019-present HiveMQ GmbH
+ *
+ *  Licensed under the Apache License, Version 2.0 (the "License");
+ *  you may not use this file except in compliance with the License.
+ *  You may obtain a copy of the License at
+ *
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ *
+ *  Unless required by applicable law or agreed to in writing, software
+ *  distributed under the License is distributed on an "AS IS" BASIS,
+ *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ *  See the License for the specific language governing permissions and
+ *  limitations under the License.
+ */
+
+package com.hivemq.protocols.fsm;
+
+public class ProtocolAdapterOperator {
+}