Bugfix/problem when sending chunked messages to an endpoint

agrirouter-sdk-java-api/src/main/java/com/dke/data/agrirouter/api/env/Environment.java

@@ -1,17 +1,31 @@
 package com.dke.data.agrirouter.api.env;
 
 import com.dke.data.agrirouter.api.enums.SecuredOnboardingResponseType;
+import org.apache.commons.lang3.BooleanUtils;
 
 /** Common Environment, holds some default methods pointing to the QA. */
 public interface Environment {
 
+  /** System property to enable logging for HTTP requests. */
+  String ENABLE_HTTP_REQUEST_LOGGING = "agrirouter.sdk.enable.http.request.logging";
+
   /** Template for MQTT connections. */
   String MQTT_URL_TEMPLATE = "ssl://%s:%s";
 
   /** Link template for the secured onboarding process. */
   String SECURED_ONBOARDING_AUTHORIZATION_LINK_TEMPLATE =
       "/application/%s/authorize?response_type=%s&state=%s&redirect_uri=%s";
 
+  /**
+   * Checks whether the HTTP request logging is enabled or not.
+   *
+   * @return -
+   */
+  static boolean httpRequestLoggingEnabled() {
+    final String httpRequestLoggingEnabled = System.getProperty(ENABLE_HTTP_REQUEST_LOGGING);
+    return BooleanUtils.toBoolean(httpRequestLoggingEnabled);
+  }
+
   /**
    * Returning the AR base URL without trailing '/'.
    *

agrirouter-sdk-java-api/src/main/java/com/dke/data/agrirouter/api/service/messaging/encoding/EncodeMessageService.java

@@ -35,7 +35,7 @@ String encode(
    * @param payloadParameters -
    * @return -
    */
-  List<MessageParameterTuple> chunkAndEncode(
+  List<MessageParameterTuple> chunkAndBase64EncodeEachChunk(
       MessageHeaderParameters messageHeaderParameters,
       PayloadParameters payloadParameters,
       OnboardingResponse onboardingResponse);

agrirouter-sdk-java-impl/src/main/java/com/dke/data/agrirouter/impl/RequestFactory.java

@@ -1,6 +1,7 @@
 package com.dke.data.agrirouter.impl;
 
 import com.dke.data.agrirouter.api.enums.CertificationType;
+import com.dke.data.agrirouter.api.env.Environment;
 import com.dke.data.agrirouter.api.exception.CertificationTypeNotSupportedException;
 import com.dke.data.agrirouter.api.exception.CouldNotCreateDynamicKeyStoreException;
 import com.dke.data.agrirouter.impl.common.ssl.KeyStoreCreationService;
@@ -14,10 +15,14 @@
 import org.glassfish.jersey.client.ClientConfig;
 import org.glassfish.jersey.client.ClientProperties;
 import org.glassfish.jersey.logging.LoggingFeature;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /** Factory to encapsulate the requests against the agrirouter */
 public final class RequestFactory {
 
+  private static final Logger LOGGER = LoggerFactory.getLogger(RequestFactory.class);
+
   /** Hidden constructor. */
   private RequestFactory() {
     // NOP
@@ -36,7 +41,13 @@ public static Invocation.Builder securedRequest(
     ClientConfig clientConfig = new ClientConfig();
     KeyStore keyStore = createKeyStore(certificate, password, certificationType);
     Client client = createClient(clientConfig, keyStore, password, certificationType);
-    client.property(LoggingFeature.LOGGING_FEATURE_LOGGER_LEVEL_CLIENT, "INFO");
+    if (Environment.httpRequestLoggingEnabled()) {
+      client.property(LoggingFeature.LOGGING_FEATURE_LOGGER_LEVEL_CLIENT, "INFO");
+    } else {
+      LOGGER.debug(
+          "Request logging is currently disabled. If you want to enable it, please set '{}'.",
+          Environment.ENABLE_HTTP_REQUEST_LOGGING);
+    }
     WebTarget target = client.target(url);
     Invocation.Builder request = target.request(MediaType.APPLICATION_JSON_TYPE);
     request.accept(MediaType.APPLICATION_JSON_TYPE);
@@ -145,7 +156,7 @@ public static Invocation.Builder bearerTokenRequest(
     return request;
   }
 
-  public class AgrirouterHttpHeader {
+  public static class AgrirouterHttpHeader {
     public static final String APPLICATION_ID = "X-Agrirouter-ApplicationId";
     public static final String SIGNATURE = "X-Agrirouter-Signature";
   }

agrirouter-sdk-java-impl/src/main/java/com/dke/data/agrirouter/impl/messaging/encoding/EncodeMessageServiceImpl.java

@@ -1,5 +1,7 @@
 package com.dke.data.agrirouter.impl.messaging.encoding;
 
+import static com.dke.data.agrirouter.api.service.parameters.PayloadParametersKt.MAX_LENGTH_FOR_RAW_MESSAGE_CONTENT;
+
 import agrirouter.commons.Chunk;
 import agrirouter.request.Request;
 import com.dke.data.agrirouter.api.dto.onboard.OnboardingResponse;
@@ -13,16 +15,12 @@
 import com.dke.data.agrirouter.impl.NonEnvironmentalService;
 import com.dke.data.agrirouter.impl.common.MessageIdService;
 import com.dke.data.agrirouter.impl.messaging.SequenceNumberService;
-import com.google.common.base.Splitter;
 import com.google.protobuf.Any;
 import com.google.protobuf.ByteString;
 import java.io.ByteArrayOutputStream;
 import java.io.IOException;
 import java.nio.charset.StandardCharsets;
-import java.util.ArrayList;
-import java.util.Base64;
-import java.util.Collections;
-import java.util.List;
+import java.util.*;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.stream.Collectors;
 import org.apache.commons.lang3.StringUtils;
@@ -92,7 +90,7 @@ public List<String> encode(List<MessageParameterTuple> messageParameterTuples) {
    * @param payloadParameters Content of the message. It shall not be Base64 encoded before.
    * @return -
    */
-  public List<MessageParameterTuple> chunkAndEncode(
+  public List<MessageParameterTuple> chunkAndBase64EncodeEachChunk(
       MessageHeaderParameters messageHeaderParameters,
       PayloadParameters payloadParameters,
       OnboardingResponse onboardingResponse) {
@@ -114,10 +112,8 @@ public List<MessageParameterTuple> chunkAndEncode(
             .debug(
                 "The message should be chunked, current size of the payload ({}) is above the limitation.",
                 payloadParameters.getValue().toStringUtf8().length());
-        String wholeMessage = payloadParameters.getValue().toStringUtf8();
-        final List<String> messageChunks =
-            Splitter.fixedLength(payloadParameters.maxLengthForMessages())
-                .splitToList(wholeMessage);
+        byte[] wholeMessage = payloadParameters.getValue().toByteArray();
+        final List<byte[]> messageChunks = splitIntoChunks(wholeMessage);
         List<MessageParameterTuple> tuples = new ArrayList<>();
         AtomicInteger chunkNr = new AtomicInteger(1);
         final String chunkContextId = ChunkContextIdService.generateChunkContextId();
@@ -135,14 +131,12 @@ public List<MessageParameterTuple> chunkAndEncode(
               chunkInfo.setContextId(chunkContextId);
               chunkInfo.setCurrent(chunkNr.get());
               chunkInfo.setTotal(messageChunks.size());
-              chunkInfo.setTotalSize(wholeMessage.length());
+              chunkInfo.setTotalSize(wholeMessage.length);
               header.setChunkInfo(chunkInfo.build());
 
               final PayloadParameters payload = new PayloadParameters();
               payload.copyFrom(payloadParameters);
-              payload.setValue(
-                  ByteString.copyFromUtf8(
-                      Base64.getEncoder().encodeToString(chunk.getBytes(StandardCharsets.UTF_8))));
+              payload.setValue(ByteString.copyFromUtf8(Base64.getEncoder().encodeToString(chunk)));
 
               tuples.add(new MessageParameterTuple(header, payload));
 
@@ -179,6 +173,23 @@ public List<MessageParameterTuple> chunkAndEncode(
     }
   }
 
+  private List<byte[]> splitIntoChunks(byte[] wholeMessage) {
+    List<byte[]> chunks = new ArrayList<>();
+    byte[] remainingBytes = wholeMessage;
+    do {
+      final byte[] chunk =
+          Arrays.copyOfRange(remainingBytes, 0, MAX_LENGTH_FOR_RAW_MESSAGE_CONTENT);
+      chunks.add(chunk);
+      remainingBytes =
+          Arrays.copyOfRange(
+              remainingBytes, MAX_LENGTH_FOR_RAW_MESSAGE_CONTENT + 1, remainingBytes.length - 1);
+    } while (remainingBytes.length > MAX_LENGTH_FOR_RAW_MESSAGE_CONTENT);
+    if (remainingBytes.length > 0) {
+      chunks.add(remainingBytes);
+    }
+    return chunks;
+  }
+
   private Request.RequestEnvelope header(MessageHeaderParameters parameters) {
     logMethodBegin(parameters);
 

agrirouter-sdk-java-impl/src/test/java/com/dke/data/agrirouter/impl/messaging/encoding/EncodeMessageServiceImplTest.java

@@ -31,7 +31,7 @@ void givenEmptyMessageWhenChunkingThenTheImplementationShouldReturnTheRightNumbe
     PayloadParameters payloadParameters = getPayloadParameters(toSendMessage);
 
     final List<MessageParameterTuple> chunks =
-        encodeMessageService.chunkAndEncode(
+        encodeMessageService.chunkAndBase64EncodeEachChunk(
             messageHeaderParameters, payloadParameters, fakeOnboardingResponse());
     Assertions.assertEquals(1, chunks.size());
   }
@@ -47,7 +47,7 @@ void givenEmptyMessageWhenChunkingThenTheImplementationShouldReturnTheRightNumbe
     PayloadParameters payloadParameters = getPayloadParameters(toSendMessage);
 
     final List<MessageParameterTuple> chunks =
-        encodeMessageService.chunkAndEncode(
+        encodeMessageService.chunkAndBase64EncodeEachChunk(
             messageHeaderParameters, payloadParameters, fakeOnboardingResponse());
     Assertions.assertEquals(1, chunks.size());
   }
@@ -66,7 +66,7 @@ void givenEmptyMessageWhenChunkingThenTheImplementationShouldReturnTheRightNumbe
     PayloadParameters payloadParameters = getPayloadParameters(toSendMessage);
 
     final List<MessageParameterTuple> chunks =
-        encodeMessageService.chunkAndEncode(
+        encodeMessageService.chunkAndBase64EncodeEachChunk(
             messageHeaderParameters, payloadParameters, fakeOnboardingResponse());
     Assertions.assertEquals(1, chunks.size());
   }
@@ -83,7 +83,7 @@ void givenEmptyMessageWhenChunkingThenTheImplementationShouldReturnTheRightNumbe
     PayloadParameters payloadParameters = getPayloadParameters(toSendMessage);
 
     final List<MessageParameterTuple> chunks =
-        encodeMessageService.chunkAndEncode(
+        encodeMessageService.chunkAndBase64EncodeEachChunk(
             messageHeaderParameters, payloadParameters, fakeOnboardingResponse());
     Assertions.assertEquals(2, chunks.size());
 

agrirouter-sdk-java-tests/src/test/java/com/dke/data/agrirouter/test/helper/ContentReader.java

@@ -16,6 +16,11 @@ public static String readBase64EncodedMessageContent(Identifier identifier) thro
     return new String(Base64.getEncoder().encode(rawData));
   }
 
+  public static byte[] readRawData(Identifier identifier) throws Throwable {
+    Path path = Paths.get(FOLDER.concat(identifier.getFileName()));
+    return Files.readAllBytes(path);
+  }
+
   public enum Identifier {
     BIG_TASK_DATA("big_taskdata.zip"),
     SMALL_TASK_DATA("small_taskdata.zip");

agrirouter-sdk-java-tests/src/test/java/com/dke/data/agrirouter/test/messaging/rest/SendChunkedMessageTest.java

@@ -20,28 +20,22 @@
 import com.dke.data.agrirouter.test.AbstractIntegrationTest;
 import com.dke.data.agrirouter.test.Assertions;
 import com.dke.data.agrirouter.test.OnboardingResponseRepository;
+import com.dke.data.agrirouter.test.helper.ContentReader;
 import com.google.protobuf.ByteString;
-import java.io.IOException;
 import java.util.*;
 import java.util.concurrent.atomic.AtomicReference;
-import java.util.stream.Stream;
-import org.apache.commons.lang3.RandomStringUtils;
 import org.apache.http.HttpStatus;
-import org.junit.jupiter.params.ParameterizedTest;
-import org.junit.jupiter.params.provider.Arguments;
-import org.junit.jupiter.params.provider.MethodSource;
+import org.junit.jupiter.api.Test;
 
 /** Test case to show the behavior for chunked message sending. */
 class SendChunkedMessageTest extends AbstractIntegrationTest {
 
   private static final int MAX_CHUNK_SIZE = 1024000;
 
-  @ParameterizedTest
-  @MethodSource("fakeRawMessageContentThatHasToBeChunked")
+  @Test
   void
-      givenLargeContentMessageWhenSendingTheMessageToTheAgrirouterTheSdkShouldHelpToSendTheFileInMultipleChunks(
-          ByteString messageContent, int expectedNrOfChunks)
-          throws IOException, InterruptedException {
+      givenLargeContentMessageWhenSendingTheMessageToTheAgrirouterTheSdkShouldHelpToSendTheFileInMultipleChunks()
+          throws Throwable {
 
     final EncodeMessageService encodeMessageService = new EncodeMessageServiceImpl();
     final SendMessageServiceImpl sendMessageService = new SendMessageServiceImpl();
@@ -55,17 +49,15 @@ class SendChunkedMessageTest extends AbstractIntegrationTest {
         SequenceNumberService.generateSequenceNumberForEndpoint(onboardingResponse));
     messageHeaderParameters.setMode(Request.RequestEnvelope.Mode.DIRECT);
     messageHeaderParameters.setRecipients(
-        Collections.singletonList(
-            OnboardingResponseRepository.read(
-                    OnboardingResponseRepository.Identifier.COMMUNICATION_UNIT)
-                .getSensorAlternateId()));
+        Collections.singletonList("37cd61d1-76eb-4145-a735-c938d05a32d8"));
 
     PayloadParameters payloadParameters = new PayloadParameters();
-    payloadParameters.setValue(messageContent);
+    payloadParameters.setValue(
+        ByteString.copyFrom(ContentReader.readRawData(ContentReader.Identifier.BIG_TASK_DATA)));
     payloadParameters.setTypeUrl(SystemMessageType.EMPTY.getKey());
 
     List<MessageParameterTuple> tuples =
-        encodeMessageService.chunkAndEncode(
+        encodeMessageService.chunkAndBase64EncodeEachChunk(
             messageHeaderParameters, payloadParameters, onboardingResponse);
 
     tuples.forEach(
@@ -92,7 +84,7 @@ class SendChunkedMessageTest extends AbstractIntegrationTest {
             new DefaultCancellationToken(MAX_TRIES_BEFORE_FAILURE, DEFAULT_INTERVAL));
 
     Assertions.assertTrue(fetchMessageResponses.isPresent());
-    Assertions.assertEquals(expectedNrOfChunks, fetchMessageResponses.get().size());
+    Assertions.assertEquals(3, fetchMessageResponses.get().size());
 
     DecodeMessageService decodeMessageService = new DecodeMessageServiceImpl();
     AtomicReference<DecodeMessageResponse> decodeMessageResponse = new AtomicReference<>();
@@ -108,28 +100,4 @@ class SendChunkedMessageTest extends AbstractIntegrationTest {
                   decodeMessageResponse.get().getResponseEnvelope().getResponseCode());
             });
   }
-
-  /**
-   * Delivers fake message content for multiple test cases.
-   *
-   * @return -
-   */
-  private static Stream<Arguments> fakeRawMessageContentThatHasToBeChunked() {
-    return Stream.of(
-        Arguments.of(
-            ByteString.copyFromUtf8(
-                RandomStringUtils.randomAlphabetic(
-                    PayloadParametersKt.MAX_LENGTH_FOR_RAW_MESSAGE_CONTENT * 3)),
-            3),
-        Arguments.of(
-            ByteString.copyFromUtf8(
-                RandomStringUtils.randomAlphabetic(
-                    PayloadParametersKt.MAX_LENGTH_FOR_RAW_MESSAGE_CONTENT * 2)),
-            2),
-        Arguments.of(
-            ByteString.copyFromUtf8(
-                RandomStringUtils.randomAlphabetic(
-                    PayloadParametersKt.MAX_LENGTH_FOR_RAW_MESSAGE_CONTENT)),
-            1));
-  }
 }