How to Build a File-based Protocol Adapter for HiveMQ Edge

git clone https://github.com/hivemq/hivemq-hello-world-protocol-adapter

@Override
    public void poll(@NotNull PollingInput pollingInput, @NotNull PollingOutput pollingOutput) {
        // here the sampling must be done. F.e. sending a http request
        pollingOutput.addDataPoint("dataPoint1", 42);
        pollingOutput.addDataPoint("dataPoint1", 1337);
        pollingOutput.finish();
    }

@Override
    public void poll(final @NotNull PollingInput<HelloWorldPollingContext> pollingInput, final @NotNull PollingOutput pollingOutput) {
        // absolute path to the file that contains the data. Magic string for now. Later it will be part of the config
        final String absolutePathToFle = "/tmp/sensor1.txt";
        try {
            final Path path = Path.of(absolutePathToFle);
            final long length = path.toFile().length();
            final int limit = 64_000; // not a constant to have a more compact code example
            if (length > limit) {
                pollingOutput.fail(String.format("File '%s' of size '%d' exceeds the limit '%d'.", path.toAbsolutePath(), length, limit));
                return;
            }
            // load the content of the file
            byte[] fileContent = Files.readAllBytes(path);
            // encode it as base64
            final String encodedFileContent = Base64.getEncoder().encodeToString(fileContent);
            // add the content of the file to the output
            pollingOutput.addDataPoint("value", encodedFileContent);
        } catch (IOException e) {
            // in case something goes wrong while reading the file, an IOException will be thrown.
            // we handle it by failing the poll process and returning from the poll method.
            pollingOutput.fail(e, "An exception occurred while reading the file '" + absolutePathToFle + "'.");
            return;
        }
        // we need to tell edge that the polling is done as edge also supports asynchronous polling.
        pollingOutput.finish();
    }

mqtt sub -t '#'

{
  "timestamp" : 1719385976555,
  "value" : "SGl2ZU1RIEVkZ2UK"
}
{
  "timestamp" : 1719385977420,
  "value" : "SGl2ZU1RIEVkZ2UK"
}
{
  "timestamp" : 1719385977555,
  "value" : "SGl2ZU1RIEVkZ2UK"
}

@JsonProperty(value = "filePath", required = true)
    @ModuleConfigField(title = "The file path",
            description = "The path to the file that should be scraped.",
            required = true)
    protected @NotNull String filePath;

    @JsonCreator
    public HelloWorldPollingContext(
            @JsonProperty("destination") @Nullable final String destination,
            @JsonProperty("qos") final int qos,
            @JsonProperty("userProperties") @Nullable List<UserProperty> userProperties,
            @JsonProperty("filePath") @NotNull String filePath) {
        this.destination = destination;
        this.qos = qos;
        if (userProperties != null) {
            this.userProperties = userProperties;
        }
        this.filePath = filePath;
    }


    public @NotNull String getFilePath() {
        return filePath;
    }

@Override
    public void poll(final @NotNull PollingInput<HelloWorldPollingContext> pollingInput, final @NotNull PollingOutput pollingOutput) {
        // absolute path to the file that contains the data 
        final String absolutePathToFle = pollingInput.getPollingContext().getFilePath();
        try {
            final Path path = Path.of(absolutePathToFle);
            final long length = path.toFile().length();
            final int limit = 64_000; // not a constant to have a more compact code example
            if (length > limit) {
                pollingOutput.fail(String.format("File '%s' of size '%d' exceeds the limit '%d'.", path.toAbsolutePath(), length, limit));
                return;
            }
            // load the content of the file
            byte[] fileContent = Files.readAllBytes(path);
            // encode it as base64
            final String encodedFileContent = Base64.getEncoder().encodeToString(fileContent);
            // add the content of the file to the output
            pollingOutput.addDataPoint("value", encodedFileContent);
        } catch (IOException e) {
            // in case something goes wrong while reading the file, an IOException will be thrown.
            // we handle it by failing the poll process and returning from the poll method.
            pollingOutput.fail(e, "An exception occurred while reading the file '" + absolutePathToFle + "'.");
            return;
        }
        // we need to tell edge that the polling is done as edge also supports asynchronous polling.
        pollingOutput.finish();
    }

@TempDir
    @NotNull File temporaryDir;     // @TempDir creates a temporary folder which will automatically be removed after the test. 

    // we mock these objects to easily control their behavior 
    private final @NotNull ProtocolAdapterInput<HelloWorldAdapterConfig> adapterInput = mock();
    private final @NotNull HelloWorldAdapterConfig config = mock();
    private final @NotNull PollingInput<HelloWorldPollingContext> pollingInput = mock();

    @Test
    void test_poll_whenFileIsPresent_thenFileContentsAreSetInOutput() throws IOException {
        final File fileWithData = new File(temporaryDir, "data.txt"); // create a temporary file which content gets polled
        Files.write(fileWithData.toPath(), "Hello World".getBytes(StandardCharsets.UTF_8)); // write "Hello World" into the file
        when(adapterInput.getConfig()).thenReturn(config); // whenever the getConfig() is called on the input object, our mocked config is returned
        when(pollingInput.getPollingContext()).thenReturn(new HelloWorldPollingContext("mqttTopic", 1, List.of(), fileWithData.getAbsolutePath()));
        // whenever the pollContext is accessed on the pollingInput, a crafted context is returned, so that our file is polled and the correct mqtt topic is set
        TestPollingOutput pollingOutput = new TestPollingOutput(); // test implementation for a pollingOutput
        HelloWorldPollingProtocolAdapter adapter = new HelloWorldPollingProtocolAdapter(new HelloWorldProtocolAdapterInformation(), adapterInput);
        // we create the adapter with the setup we want for the test

        adapter.poll(pollingInput, pollingOutput); // make the adapter poll the file and insert the data into the output object. 
        // No further sync is needed here as the poll method is not asynchronous in the example

        final Object value = pollingOutput.getDataPoints().get("value"); // get the data point from the output
        assertNotNull(value); // check that there is a value present
        assertTrue(value instanceof String); // check that it has the expected type String
        String valueAsString = (String) value; // cast it to String
        final byte[] decodedBytes = Base64.getDecoder().decode(valueAsString); // decode it as the value is encoded as Base64
        final String actualDecodedValue = new String(decodedBytes, StandardCharsets.UTF_8); // create a UTF-8 String of it
        assertEquals("Hello World", actualDecodedValue); // check that the string is correct
    }

{
  "ui:tabs": [
    {
      "id": "coreFields",
      "title": "Core Fields",
      "properties": [
        "id"
      ]
    },
    {
      "id": "subFields",
      "title": "Subscription",
      "properties": [
        "subscriptions"
      ]
    },
    {
      "id": "publishing",
      "title": "protocolAdapter.uiSchema.groups.publishing",
      "properties": [
        "maxPollingErrorsBeforeRemoval",
        "publishChangedDataOnly",
        "pollingIntervalMillis"
      ]
    }
  ],
  "subscriptions": {
    "ui:batchMode": true,
    "items": {
      "ui:order": [
        "destination",
        "*",
        "userProperties"
      ],
      "ui:collapsable": {
        "titleKey": "destination"
      }
    }
  }
}

public class HelloWorldProtocolAdapterInformation
 implements ProtocolAdapterInformation {

    public static final @NotNull ProtocolAdapterInformation INSTANCE = new HelloWorldProtocolAdapterInformation();

    protected HelloWorldProtocolAdapterInformation() {
    }

    @Override
    public @NotNull String getProtocolName() {
        // the returned string will be used for logging information on the protocol adapter
        return "PollingFileProtocol";
    }

    @Override
    public @NotNull String getProtocolId() {
        // this id is very important as this is how the adapters configurations in the config.xml are linked to the adapter implementations.
        // any change here means you will need to edit the config.xml
        return "Polling_File_Protocol";
    }

    @Override
    public @NotNull String getDisplayName() {
        // the name for this protocol adapter type that will be displayed within edge's ui
        return "Polling File Protocol Adapter";
    }

    @Override
    public @NotNull String getDescription() {
        // the description that will be shown for this protocol adapter within edge's ui
        return "This Protocol Adapter periodically polls and publishes the content of a given file.";
    }

    @Override
    public @NotNull String getUrl() {
        // this url will be displayed in the ui as a link to further documentation on this protocol adapter.
        // e.g. this could be a link to the source code and a readme
        return "https://www.hivemq.com/";
    }

    @Override
    public @NotNull String getVersion() {
        // the version of this protocol adapter, the usage of semantic versioning is advised.
        return "0.1.0";
    }

    @Override
    public @NotNull EnumSet<ProtocolAdapterCapability> getCapabilities() {
        // this indicates what capabilities this protocol adapter has. E.g. READ/WRITE. See the ProtocolAdapterCapability enum for more information.
        return EnumSet.of(ProtocolAdapterCapability.READ);
    }

    @Override
    public @NotNull String getLogoUrl() {
        // this is a default image that is always available.
        return "/images/helloWorld.png";
    }

    @Override
    public @NotNull String getAuthor() {
        return "HiveMQ";
    }

    @Override
    public @Nullable ProtocolAdapterCategory getCategory() {
        // this indicates for which use cases this protocol adapter is intended. See the ProtocolAdapterConstants.CATEGORY enum for more information.
        return ProtocolAdapterCategory.CONNECTIVITY;
    }

    @Override
    public List<ProtocolAdapterTag> getTags() {
        // here you can set which Tags should be applied to this protocol adapter
        return List.of(ProtocolAdapterTag.AUTOMATION);
    }
}