improve handling of native types

Author: DC2-DanielKrueger

modules/hivemq-edge-module-postgresql/src/main/java/com/hivemq/edge/adapters/postgresql/PostgreSQLPollingProtocolAdapter.java

@@ -40,6 +40,8 @@
 import java.sql.PreparedStatement;
 import java.sql.ResultSet;
 import java.sql.ResultSetMetaData;
+import java.sql.SQLException;
+import java.sql.Types;
 import java.util.ArrayList;
 import java.util.List;
 
@@ -140,48 +142,77 @@ public void poll(final @NotNull PollingInput pollingInput, final @NotNull Pollin
     }
 
     private void loadDataFromDB(final @NotNull PollingOutput output, final @NotNull PostgreSQLAdapterTag tag) {
-        try {
+        // ARM to ensure the connection is closed afterwards
+        try (final Connection connection = databaseConnection.getConnection()) {
             log.debug("Getting tag definition");
             /* Get the tag definition (Query, RowLimit and Split Lines)*/
             final PostgreSQLAdapterTagDefinition definition = tag.getDefinition();
 
             /* Execute query and handle result */
-            try (final Connection connection = databaseConnection.getConnection()) {
-                final PreparedStatement preparedStatement = connection.prepareStatement(tag.getDefinition().getQuery());
-                final ResultSet result = preparedStatement.executeQuery();
-                assert result != null;
-                final ArrayList<ObjectNode> resultObject = new ArrayList<>();
-                final ResultSetMetaData resultSetMD = result.getMetaData();
-                while (result.next()) {
-                    final int numColumns = resultSetMD.getColumnCount();
-                    final ObjectNode node = OBJECT_MAPPER.createObjectNode();
-                    for (int i = 1; i <= numColumns; i++) {
-                        final String column_name = resultSetMD.getColumnName(i);
-                        node.put(column_name, result.getString(column_name));
-                    }
-
-                    /* Publish datapoint with a single line if split is required */
-                    if (definition.getSpiltLinesInIndividualMessages()) {
-                        log.debug("Splitting lines in multiple messages");
-                        output.addDataPoint("queryResult", node);
-                    } else {
-                        resultObject.add(node);
-                    }
+
+            final PreparedStatement preparedStatement = connection.prepareStatement(tag.getDefinition().getQuery());
+            final ResultSet result = preparedStatement.executeQuery();
+            assert result != null;
+            final ArrayList<ObjectNode> resultObject = new ArrayList<>();
+            final ResultSetMetaData resultSetMD = result.getMetaData();
+            while (result.next()) {
+                final int numColumns = resultSetMD.getColumnCount();
+                final ObjectNode node = OBJECT_MAPPER.createObjectNode();
+                for (int i = 1; i <= numColumns; i++) {
+                    parseAndAddValue(i, result, resultSetMD, node);
                 }
 
-                /* Publish datapoint with all lines if no split is required */
-                if (!definition.getSpiltLinesInIndividualMessages()) {
-                    log.debug("Publishing all lines in a single message");
-                    output.addDataPoint("queryResult", resultObject);
+                /* Publish datapoint with a single line if split is required */
+                if (definition.getSpiltLinesInIndividualMessages()) {
+                    log.debug("Splitting lines in multiple messages");
+                    output.addDataPoint("queryResult", node);
+                } else {
+                    resultObject.add(node);
                 }
-            } catch (final Exception e) {
-                output.fail(e, null);
+            }
+
+            /* Publish datapoint with all lines if no split is required */
+            if (!definition.getSpiltLinesInIndividualMessages()) {
+                log.debug("Publishing all lines in a single message");
+                output.addDataPoint("queryResult", resultObject);
             }
         } catch (final Exception e) {
             output.fail(e, null);
         }
     }
 
+    // according to https://www.ibm.com/docs/en/db2/11.1?topic=djr-sql-data-type-representation
+    private void parseAndAddValue(
+            final int index,
+            final @NotNull ResultSet result,
+            final @NotNull ResultSetMetaData resultSetMD,
+            final @NotNull ObjectNode node) throws SQLException {
+        final String columnName = resultSetMD.getColumnName(index);
+        final int columnType = resultSetMD.getColumnType(index);
+        switch (columnType) {
+            case Types.BIT:
+            case Types.TINYINT:
+            case Types.SMALLINT:
+            case Types.INTEGER:
+                node.put(columnName, result.getInt(index));
+                return;
+            case Types.BIGINT:
+                node.put(columnName, result.getLong(index));
+                return;
+            case Types.DECIMAL:
+                node.put(columnName, result.getBigDecimal(index));
+                return;
+            case Types.REAL:
+            case Types.FLOAT:
+            case Types.DOUBLE:
+            case Types.NUMERIC:
+                node.put(columnName, result.getDouble(index));
+                return;
+            default:
+                node.put(columnName, result.getString(index));
+        }
+    }
+
     @Override
     public int getPollingIntervalMillis() {
         return adapterConfig.getPollingIntervalMillis();