add generic type to PollingInput for the PollingContext (#445)

* add generic type to PollingInput for the PollingContext

* fix compilation

Author: DC2-DanielKrueger

hivemq-edge/src/main/java/com/hivemq/edge/modules/adapters/simulation/SimulationAdapterConfig.java

@@ -19,8 +19,8 @@
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.fasterxml.jackson.annotation.JsonPropertyOrder;
 import com.hivemq.adapter.sdk.api.annotations.ModuleConfigField;
+import com.hivemq.adapter.sdk.api.config.PollingContext;
 import com.hivemq.adapter.sdk.api.config.ProtocolAdapterConfig;
-import com.hivemq.edge.modules.config.impl.PollingContextImpl;
 import com.hivemq.extension.sdk.api.annotations.NotNull;
 
 import java.util.ArrayList;
@@ -69,7 +69,7 @@ public int getMaxPollingErrorsBeforeRemoval() {
     @ModuleConfigField(title = "Subscriptions",
                        description = "List of subscriptions for the simulation",
                        required = true)
-    private @NotNull List<PollingContextImpl> adapterSubscriptions = new ArrayList<>();
+    private @NotNull List<PollingContext> adapterSubscriptions = new ArrayList<>();
 
     @JsonProperty("minValue")
     @ModuleConfigField(title = "Min. Generated Value",
@@ -89,7 +89,7 @@ public SimulationAdapterConfig() {
     }
 
     public SimulationAdapterConfig(
-            final @NotNull String id, final @NotNull List<PollingContextImpl> adapterSubscriptions) {
+            final @NotNull String id, final @NotNull List<PollingContext> adapterSubscriptions) {
         this.id = id;
         this.adapterSubscriptions = adapterSubscriptions;
     }
@@ -103,11 +103,11 @@ public void setId(final @NotNull String id) {
     }
 
 
-    public void setSubscriptions(List<PollingContextImpl> adapterSubscriptions) {
+    public void setSubscriptions(@NotNull List<PollingContext> adapterSubscriptions) {
         this.adapterSubscriptions = adapterSubscriptions;
     }
 
-    public @NotNull List<PollingContextImpl> getSubscriptions() {
+    public @NotNull List<PollingContext> getSubscriptions() {
         return adapterSubscriptions;
     }
 

hivemq-edge/src/main/java/com/hivemq/edge/modules/adapters/simulation/SimulationProtocolAdapter.java

@@ -33,7 +33,7 @@
 
 import static com.hivemq.adapter.sdk.api.state.ProtocolAdapterState.ConnectionStatus.STATELESS;
 
-public class SimulationProtocolAdapter implements PollingProtocolAdapter {
+public class SimulationProtocolAdapter implements PollingProtocolAdapter<PollingContext> {
 
     private final @NotNull ProtocolAdapterInformation adapterInformation;
     private final @NotNull SimulationAdapterConfig adapterConfig;
@@ -71,7 +71,7 @@ public void stop(@NotNull final ProtocolAdapterStopInput input, @NotNull final P
 
     @Override
     public void poll(
-            final @NotNull PollingInput pollingInput, final @NotNull PollingOutput pollingOutput) {
+            final @NotNull PollingInput<com.hivemq.adapter.sdk.api.config.PollingContext> pollingInput, final @NotNull PollingOutput pollingOutput) {
         pollingOutput.addDataPoint("sample",
                 ThreadLocalRandom.current()
                         .nextDouble(Math.min(adapterConfig.getMinValue(), adapterConfig.getMaxValue()),
@@ -80,7 +80,7 @@ public void poll(
     }
 
     @Override
-    public @NotNull List<? extends PollingContext> getPollingContexts() {
+    public @NotNull List<PollingContext> getPollingContexts() {
         return adapterConfig.getSubscriptions();
     }
 

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

@@ -14,20 +14,20 @@
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.TimeUnit;
 
-public class PerSubscriptionSampler extends AbstractSubscriptionSampler {
+public class PerSubscriptionSampler<T extends PollingContext> extends AbstractSubscriptionSampler {
 
     private static final Logger log = LoggerFactory.getLogger(PerSubscriptionSampler.class);
 
 
     private final @NotNull PollingProtocolAdapter perSubscriptionProtocolAdapter;
-    private final @NotNull PollingContext pollingContext;
+    private final @NotNull T pollingContext;
 
     public PerSubscriptionSampler(
             final @NotNull ProtocolAdapterWrapper<PollingProtocolAdapter> protocolAdapter,
             final @NotNull MetricRegistry metricRegistry,
             final @NotNull ObjectMapper objectMapper,
             final @NotNull ProtocolAdapterPublishService adapterPublishService,
-            final @NotNull PollingContext pollingContext,
+            final @NotNull T pollingContext,
             final @NotNull EventService eventService) {
         super(protocolAdapter,
                 protocolAdapter.getAdapter().getPollingIntervalMillis(),
@@ -48,7 +48,7 @@ public PerSubscriptionSampler(
         }
         final PollingOutputImpl pollingOutput = new PollingOutputImpl(new ProtocolAdapterDataSampleImpl());
         try {
-            perSubscriptionProtocolAdapter.poll(new PollingInputImpl(pollingContext), pollingOutput);
+            perSubscriptionProtocolAdapter.poll(new PollingInputImpl<>(pollingContext), pollingOutput);
         } catch (Throwable t) {
             pollingOutput.fail(t, null);
             throw t;

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

@@ -4,16 +4,16 @@
 import com.hivemq.adapter.sdk.api.polling.PollingInput;
 import com.hivemq.extension.sdk.api.annotations.NotNull;
 
-public class PollingInputImpl implements PollingInput {
+public class PollingInputImpl<T extends  PollingContext> implements PollingInput<T> {
 
-    private final @NotNull PollingContext pollingContext;
+    private final @NotNull T pollingContext;
 
-    public PollingInputImpl(final @NotNull PollingContext pollingContext) {
+    public PollingInputImpl(final @NotNull T pollingContext) {
         this.pollingContext = pollingContext;
     }
 
     @Override
-    public @NotNull PollingContext getPollingContext() {
+    public @NotNull T getPollingContext() {
         return pollingContext;
     }
 }

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

@@ -23,6 +23,7 @@
 import com.google.common.util.concurrent.ListenableFuture;
 import com.hivemq.adapter.sdk.api.ProtocolAdapter;
 import com.hivemq.adapter.sdk.api.ProtocolAdapterInformation;
+import com.hivemq.adapter.sdk.api.config.PollingContext;
 import com.hivemq.adapter.sdk.api.config.ProtocolAdapterConfig;
 import com.hivemq.adapter.sdk.api.events.EventService;
 import com.hivemq.adapter.sdk.api.events.model.Event;
@@ -268,14 +269,13 @@ private void schedulePolling(final @NotNull ProtocolAdapterWrapper protocolAdapt
         if (protocolAdapterWrapper.getAdapter() instanceof PollingProtocolAdapter) {
             log.info("Scheduling polling for adapter {}", protocolAdapterWrapper.getId());
             final PollingProtocolAdapter adapter = (PollingProtocolAdapter) protocolAdapterWrapper.getAdapter();
-
             adapter.getPollingContexts().forEach(adapterSubscription -> {
                 //noinspection unchecked this is safe as we literally make a check on the adapter class before
-                final PerSubscriptionSampler sampler = new PerSubscriptionSampler(protocolAdapterWrapper,
+                final PerSubscriptionSampler<? extends PollingContext> sampler = new PerSubscriptionSampler<>(protocolAdapterWrapper,
                         metricRegistry,
                         objectMapper,
                         moduleServices.adapterPublishService(),
-                        adapterSubscription,
+                        (PollingContext)adapterSubscription,
                         eventService);
                 protocolAdapterPollingService.schedulePolling(protocolAdapterWrapper, sampler);
             });

hivemq-edge/src/test/java/com/hivemq/configuration/writer/LargeConfigFileTest.java

@@ -19,6 +19,7 @@
 import com.fasterxml.jackson.annotation.JsonProperty;
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.hivemq.adapter.sdk.api.annotations.ModuleConfigField;
+import com.hivemq.adapter.sdk.api.config.PollingContext;
 import com.hivemq.configuration.entity.HiveMQConfigEntity;
 import com.hivemq.configuration.reader.ConfigFileReaderWriter;
 import com.hivemq.configuration.reader.ConfigurationFile;
@@ -62,7 +63,7 @@ public void readSmallWriteLarge_reread_config() throws IOException {
         start = printTimer("Initial Read", System.out, start);
 
         SimulationAdapterConfig config = readConfig(mapper, entity.getProtocolAdapterConfig());
-        List<PollingContextImpl> subscriptions  = config.getSubscriptions();
+        List<PollingContext> subscriptions  = config.getSubscriptions();
 
         for (int i = 0; i < 100000; i++){
             subscriptions.add(new PollingContextImpl("foo" + i, 1, null));

modules/hivemq-edge-module-modbus/src/main/java/com/hivemq/edge/adapters/modbus/ModbusProtocolAdapter.java

@@ -48,7 +48,7 @@
 
 import static com.hivemq.adapter.sdk.api.state.ProtocolAdapterState.ConnectionStatus.CONNECTED;
 
-public class ModbusProtocolAdapter implements PollingProtocolAdapter {
+public class ModbusProtocolAdapter implements PollingProtocolAdapter<ModbusAdapterConfig.PollingContextImpl> {
     private static final Logger log = LoggerFactory.getLogger(ModbusProtocolAdapter.class);
     private final @NotNull Object lock = new Object();
     private final @NotNull ProtocolAdapterInformation adapterInformation;
@@ -96,7 +96,7 @@ public void stop(@NotNull final ProtocolAdapterStopInput input, @NotNull final P
 
     @Override
     public void poll(
-            final @NotNull PollingInput pollingInput, final @NotNull PollingOutput pollingOutput) {
+            final @NotNull PollingInput<ModbusAdapterConfig.PollingContextImpl> pollingInput, final @NotNull PollingOutput pollingOutput) {
 
         //-- If a previously linked job has terminally disconnected the client
         //-- we need to ensure any orphaned jobs tidy themselves up properly
@@ -122,7 +122,7 @@ public void poll(
     }
 
     @Override
-    public @NotNull List<? extends PollingContext> getPollingContexts() {
+    public @NotNull List<ModbusAdapterConfig.PollingContextImpl> getPollingContexts() {
         return new ArrayList<>(adapterConfig.getSubscriptions());
     }
 

modules/hivemq-edge-module-plc4x/src/main/java/com/hivemq/edge/adapters/plc4x/impl/AbstractPlc4xAdapter.java

@@ -58,7 +58,7 @@
  *
  * @author Simon L Johnson
  */
-public abstract class AbstractPlc4xAdapter<T extends Plc4xAdapterConfig> implements PollingProtocolAdapter {
+public abstract class AbstractPlc4xAdapter<T extends Plc4xAdapterConfig, C extends Plc4xAdapterConfig.PollingContextImpl> implements PollingProtocolAdapter<C> {
 
     protected static final String TAG_ADDRESS_TYPE_SEP = ":";
     private final Logger log = LoggerFactory.getLogger(getClass());
@@ -110,9 +110,9 @@ protected void handleDataAndExceptions(
     }
 
     @Override
-    public @NotNull List<? extends PollingContext> getPollingContexts() {
+    public @NotNull List<C> getPollingContexts() {
         if (getReadType() == ReadType.Read) {
-            return adapterConfig.getSubscriptions();
+            return (List<C>) adapterConfig.getSubscriptions();
         } else {
             return List.of();
         }

modules/hivemq-edge-module-plc4x/src/main/java/com/hivemq/edge/adapters/plc4x/types/ads/ADSProtocolAdapter.java

@@ -27,7 +27,7 @@
 /**
  * @author HiveMQ Adapter Generator
  */
-public class ADSProtocolAdapter extends AbstractPlc4xAdapter<ADSAdapterConfig> {
+public class ADSProtocolAdapter extends AbstractPlc4xAdapter<ADSAdapterConfig, Plc4xAdapterConfig.PollingContextImpl> {
 
     static final String SOURCE_AMS_NET_ID = "sourceAmsNetId", SOURCE_AMS_PORT = "sourceAmsPort", TARGET_AMS_PORT =
             "targetAmsPort", TARGET_AMS_NET_ID = "targetAmsNetId";

modules/hivemq-edge-module-plc4x/src/main/java/com/hivemq/edge/adapters/plc4x/types/eip/EIPProtocolAdapter.java

@@ -31,7 +31,7 @@
 /**
  * @author HiveMQ Adapter Generator
  */
-public class EIPProtocolAdapter extends AbstractPlc4xAdapter<EIPAdapterConfig> {
+public class EIPProtocolAdapter extends AbstractPlc4xAdapter<EIPAdapterConfig, Plc4xAdapterConfig.PollingContextImpl> {
 
     static final String SLOT = "slot", BACKPLANE = "backplane";