Minor change in poll method.

The issue is about sending POST requests for committing ofsets in `poll` method
even if no messages in `poll` response.
The backend sends back an error about invalid parameters of committing offsets
requests in this case.

This fix checks the count of messages in the poll
response and sends committing requests only if messages are retrieved.

Relates-To: OLPEDGE-2663
Signed-off-by: Oleksii Zubko <[email protected]>

Author: OleksiiZubko

@here/olp-sdk-dataservice-read/lib/client/StreamLayerClient.ts

@@ -200,37 +200,43 @@ export class StreamLayerClient {
         // Update xCorrelationId
         this.xCorrelationId = consumeResponse.xCorrelationId;
 
-        // Commit offsets
-        const latestOffsets: { [key: string]: number } = consumeResponse.data
-            .map(msg => msg.offset)
-            .reduce(
-                (
-                    acc: { [key: string]: number },
-                    curr: StreamApi.StreamOffset
-                ) => {
-                    acc[curr.partition] =
-                        acc[curr.partition] > curr.offset
-                            ? acc[curr.partition]
-                            : curr.offset;
-                    return acc;
+        if (consumeResponse.data.length > 0) {
+            // Commit offsets
+            const latestOffsets: {
+                [key: string]: number;
+            } = consumeResponse.data
+                .map(msg => msg.offset)
+                .reduce(
+                    (
+                        acc: { [key: string]: number },
+                        curr: StreamApi.StreamOffset
+                    ) => {
+                        acc[curr.partition] =
+                            acc[curr.partition] > curr.offset
+                                ? acc[curr.partition]
+                                : curr.offset;
+                        return acc;
+                    },
+                    {}
+                );
+
+            await StreamApi.doCommitOffsets(requestBuilder, {
+                commitOffsets: {
+                    offsets: Object.keys(latestOffsets).map(key => ({
+                        partition: +key,
+                        offset: latestOffsets[key]
+                    }))
                 },
-                {}
-            );
-
-        await StreamApi.doCommitOffsets(requestBuilder, {
-            commitOffsets: {
-                offsets: Object.keys(latestOffsets).map(key => ({
-                    partition: +key,
-                    offset: latestOffsets[key]
-                }))
-            },
-            subscriptionId: request.getSubscriptionId(),
-            mode: request.getMode(),
-            layerId: this.layerId,
-            xCorrelationId: this.xCorrelationId
-        }).catch(async error => {
-            console.log(`Commit offsets unsuccessful, error=${error.message}`);
-        });
+                subscriptionId: request.getSubscriptionId(),
+                mode: request.getMode(),
+                layerId: this.layerId,
+                xCorrelationId: this.xCorrelationId
+            }).catch(async error => {
+                console.log(
+                    `Commit offsets unsuccessful, error=${error.message}`
+                );
+            });
+        }
 
         return Promise.resolve(consumeResponse.data);
     }

@here/olp-sdk-dataservice-read/test/unit/StreamLayerClient.test.ts

@@ -619,4 +619,109 @@ describe("StreamLayerClient", function() {
                 );
             });
     });
+
+    it("Should not post offsets if no data", async function() {
+        const headers = new Headers();
+        headers.append("X-Correlation-Id", "9141392.f96875c-9422-4df4-bdfj");
+        const mockResponse = ({
+            headers,
+            json: function() {
+                return {
+                    messages: []
+                };
+            }
+        } as unknown) as Response;
+        pollStub.callsFake(
+            (builder: any, params: any): Promise<Response> => {
+                return Promise.resolve(mockResponse);
+            }
+        );
+
+        let settings = sandbox.createStubInstance(core.OlpClientSettings);
+        const subscribtionId = await streamLayerClient.subscribe(
+            new dataServiceRead.SubscribeRequest().withMode("serial")
+        );
+        const request = new dataServiceRead.PollRequest().withSubscriptionId(
+            subscribtionId
+        );
+        const messages = await streamLayerClient.poll(request);
+
+        assert.isDefined(messages);
+        expect(messages.length).to.be.equal(0);
+        expect(commitOffsetsStub.notCalled);
+    });
+
+    it("Should not throw if commiting offsets failed", async function() {
+        const headers = new Headers();
+        headers.append("X-Correlation-Id", "9141392.f96875c-9422-4df4-bdfj");
+        const mockResponse = ({
+            headers,
+            json: function() {
+                return {
+                    messages: [
+                        {
+                            metaData: {
+                                partition: "314010583",
+                                checksum: "ff7494d6f17da702862e550c907c0a91",
+                                compressedDataSize: 152417,
+                                dataSize: 100500,
+                                data: "",
+                                dataHandle:
+                                    "iVBORw0-Lf9HdIZBfNEiKAA-AAAE-lFTkSuQmCC",
+                                timestamp: 1517916706
+                            },
+                            offset: {
+                                partition: 7,
+                                offset: 38562
+                            }
+                        },
+                        {
+                            metaData: {
+                                partition: "314010584",
+                                checksum: "ff7494d6f17da702862e550c907c0a91",
+                                dataSize: 100500,
+                                data:
+                                    "7n348c7y49nry394y39yv39y384tvn3984tvn34ty034ynt3yvt983ny",
+                                dataHandle: "",
+                                timestamp: 1517916707
+                            },
+                            offset: {
+                                partition: 8,
+                                offset: 38563
+                            }
+                        }
+                    ]
+                };
+            }
+        } as unknown) as Response;
+        pollStub.callsFake(
+            (builder: any, params: any): Promise<Response> => {
+                return Promise.resolve(mockResponse);
+            }
+        );
+
+        commitOffsetsStub.callsFake(
+            (builder: any, params: any): Promise<Response> => {
+                return Promise.reject(new Error("fake error"));
+            }
+        );
+
+        let settings = sandbox.createStubInstance(core.OlpClientSettings);
+        const params = {
+            catalogHrn: mockedHRN,
+            layerId: mockedLayerId,
+            settings: (settings as unknown) as core.OlpClientSettings
+        };
+        const subscribtionId = await streamLayerClient.subscribe(
+            new dataServiceRead.SubscribeRequest().withMode("serial")
+        );
+        const request = new dataServiceRead.PollRequest().withSubscriptionId(
+            subscribtionId
+        );
+        const messages = await streamLayerClient.poll(request);
+
+        assert.isDefined(messages);
+        expect(messages.length).to.be.equal(2);
+        expect(messages[0].metaData.dataSize).to.be.equal(100500);
+    });
 });