This is an automated email from the ASF dual-hosted git repository.
davsclaus pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new f8b8f4f2f81c CAMEL-24346: camel-google-sheets - number every range and
tolerate a range without values
f8b8f4f2f81c is described below
commit f8b8f4f2f81c025124fe64f8fc1c19e18d91652a
Author: Andrea Cosentino <[email protected]>
AuthorDate: Fri Aug 7 09:26:35 2026 +0200
CAMEL-24346: camel-google-sheets - number every range and tolerate a range
without values
Fix two defects in GoogleSheetsStreamConsumer: the splitResults range
counter was
re-created inside the loop so every exchange reported
CamelGoogleSheetsRangeIndex=1,
and a range without values threw NPE because the Sheets API omits the field
for
empty ranges. Also adds forceConsumerAsReady() to the whole-spreadsheet
code path.
Closes #25400
Co-Authored-By: Claude Opus 5 (1M context) <[email protected]>
---
.../camel-google/camel-google-sheets/pom.xml | 5 +
.../sheets/stream/GoogleSheetsStreamConsumer.java | 83 +++++++++-------
.../GoogleSheetsStreamConsumerRangeIndexTest.java | 108 +++++++++++++++++++++
3 files changed, 163 insertions(+), 33 deletions(-)
diff --git a/components/camel-google/camel-google-sheets/pom.xml
b/components/camel-google/camel-google-sheets/pom.xml
index 78bfc57f7fa7..e236158b82cb 100644
--- a/components/camel-google/camel-google-sheets/pom.xml
+++ b/components/camel-google/camel-google-sheets/pom.xml
@@ -148,6 +148,11 @@
<artifactId>camel-test-junit6</artifactId>
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.assertj</groupId>
+ <artifactId>assertj-core</artifactId>
+ <scope>test</scope>
+ </dependency>
<dependency>
<groupId>org.apache.camel</groupId>
<artifactId>camel-mock</artifactId>
diff --git
a/components/camel-google/camel-google-sheets/src/main/java/org/apache/camel/component/google/sheets/stream/GoogleSheetsStreamConsumer.java
b/components/camel-google/camel-google-sheets/src/main/java/org/apache/camel/component/google/sheets/stream/GoogleSheetsStreamConsumer.java
index f39cb20a4b3d..7201f9cd340a 100644
---
a/components/camel-google/camel-google-sheets/src/main/java/org/apache/camel/component/google/sheets/stream/GoogleSheetsStreamConsumer.java
+++
b/components/camel-google/camel-google-sheets/src/main/java/org/apache/camel/component/google/sheets/stream/GoogleSheetsStreamConsumer.java
@@ -23,6 +23,7 @@ import java.util.List;
import java.util.Queue;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;
+import java.util.stream.Stream;
import com.google.api.services.sheets.v4.Sheets;
import com.google.api.services.sheets.v4.model.BatchGetValuesResponse;
@@ -84,39 +85,7 @@ public class GoogleSheetsStreamConsumer extends
ScheduledBatchPollingConsumer {
forceConsumerAsReady();
if (response.getValueRanges() != null) {
- if (getConfiguration().isSplitResults()) {
- for (ValueRange valueRange : response.getValueRanges()) {
- AtomicInteger rangeIndex = new AtomicInteger(1);
- AtomicInteger valueIndex = new AtomicInteger();
- if (getConfiguration().getMaxResults() > 0) {
- valueRange.getValues().stream()
- .limit(getConfiguration().getMaxResults())
- .map(values ->
createExchange(rangeIndex.get(), valueIndex.incrementAndGet(),
- valueRange.getRange(),
valueRange.getMajorDimension(), values))
- .forEach(answer::add);
- } else {
- valueRange.getValues().stream()
- .map(values ->
createExchange(rangeIndex.get(), valueIndex.incrementAndGet(),
- valueRange.getRange(),
valueRange.getMajorDimension(), values))
- .forEach(answer::add);
- }
- rangeIndex.incrementAndGet();
- }
- } else {
- AtomicInteger rangeIndex = new AtomicInteger();
- response.getValueRanges()
- .stream()
- .peek(valueRange -> {
- if (getConfiguration().getMaxResults() > 0) {
- valueRange.setValues(valueRange.getValues()
- .stream()
-
.limit(getConfiguration().getMaxResults())
- .collect(Collectors.toList()));
- }
- })
- .map(valueRange ->
createExchange(rangeIndex.incrementAndGet(), valueRange))
- .forEach(answer::add);
- }
+ answer.addAll(createExchanges(response.getValueRanges()));
}
} else {
Sheets.Spreadsheets.Get request =
getClient().spreadsheets().get(getConfiguration().getSpreadsheetId());
@@ -124,12 +93,60 @@ public class GoogleSheetsStreamConsumer extends
ScheduledBatchPollingConsumer {
request.setIncludeGridData(getConfiguration().isIncludeGridData());
Spreadsheet spreadsheet = request.execute();
+
+ // okay we have some response from Google so lets mark the
consumer as ready
+ forceConsumerAsReady();
+
answer.add(createExchange(spreadsheet));
}
return processBatch(CastUtils.cast(answer));
}
+ /**
+ * Builds one exchange per value when the results are split, one per range
otherwise. The range index counts the
+ * ranges of the whole response, not of a single range.
+ */
+ Queue<Exchange> createExchanges(List<ValueRange> valueRanges) {
+ Queue<Exchange> answer = new ArrayDeque<>();
+ AtomicInteger rangeIndex = new AtomicInteger();
+
+ if (getConfiguration().isSplitResults()) {
+ for (ValueRange valueRange : valueRanges) {
+ int currentRange = rangeIndex.incrementAndGet();
+ AtomicInteger valueIndex = new AtomicInteger();
+ Stream<List<Object>> values = valuesOf(valueRange).stream();
+ if (getConfiguration().getMaxResults() > 0) {
+ values = values.limit(getConfiguration().getMaxResults());
+ }
+ values.map(value -> createExchange(currentRange,
valueIndex.incrementAndGet(),
+ valueRange.getRange(), valueRange.getMajorDimension(),
value))
+ .forEach(answer::add);
+ }
+ } else {
+ valueRanges.stream()
+ .peek(valueRange -> {
+ if (getConfiguration().getMaxResults() > 0) {
+ valueRange.setValues(valuesOf(valueRange)
+ .stream()
+ .limit(getConfiguration().getMaxResults())
+ .collect(Collectors.toList()));
+ }
+ })
+ .map(valueRange ->
createExchange(rangeIndex.incrementAndGet(), valueRange))
+ .forEach(answer::add);
+ }
+
+ return answer;
+ }
+
+ /**
+ * The values of a range, never null: the sheets API omits the field for a
range that holds no value at all.
+ */
+ private static List<List<Object>> valuesOf(ValueRange valueRange) {
+ return valueRange.getValues() != null ? valueRange.getValues() :
List.of();
+ }
+
@Override
public int processBatch(Queue<Object> exchanges) throws Exception {
int total = exchanges.size();
diff --git
a/components/camel-google/camel-google-sheets/src/test/java/org/apache/camel/component/google/sheets/stream/GoogleSheetsStreamConsumerRangeIndexTest.java
b/components/camel-google/camel-google-sheets/src/test/java/org/apache/camel/component/google/sheets/stream/GoogleSheetsStreamConsumerRangeIndexTest.java
new file mode 100644
index 000000000000..3bd796625dee
--- /dev/null
+++
b/components/camel-google/camel-google-sheets/src/test/java/org/apache/camel/component/google/sheets/stream/GoogleSheetsStreamConsumerRangeIndexTest.java
@@ -0,0 +1,108 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.google.sheets.stream;
+
+import java.util.List;
+import java.util.Queue;
+
+import com.google.api.services.sheets.v4.model.ValueRange;
+import org.apache.camel.Exchange;
+import org.apache.camel.impl.DefaultCamelContext;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Test;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatCode;
+
+/**
+ * Verifies the exchanges the stream consumer builds out of a batch-get
response: one per value when the results are
+ * split, one per range otherwise, and the range index counting the ranges of
the response.
+ */
+class GoogleSheetsStreamConsumerRangeIndexTest {
+
+ private DefaultCamelContext context;
+
+ @AfterEach
+ void tearDown() {
+ if (context != null) {
+ context.stop();
+ }
+ }
+
+ private GoogleSheetsStreamConsumer consumer(String query) throws Exception
{
+ if (context != null) {
+ context.stop();
+ }
+ context = new DefaultCamelContext();
+ // the endpoint is deliberately not started, so no google client is
built
+ GoogleSheetsStreamComponent component
+ = context.getComponent("google-sheets-stream",
GoogleSheetsStreamComponent.class);
+ GoogleSheetsStreamEndpoint endpoint = (GoogleSheetsStreamEndpoint)
component.createEndpoint(
+
"google-sheets-stream://sheet1?clientId=id&clientSecret=secret&range=A1:B2,C1:D2"
+ query);
+ return new GoogleSheetsStreamConsumer(endpoint, exchange -> {
+ });
+ }
+
+ private static ValueRange range(String name, List<List<Object>> values) {
+ return new
ValueRange().setRange(name).setMajorDimension("ROWS").setValues(values);
+ }
+
+ private static List<Object> rangeIndexes(Queue<Exchange> exchanges) {
+ return exchanges.stream()
+ .map(e ->
e.getIn().getHeader(GoogleSheetsStreamConstants.RANGE_INDEX))
+ .toList();
+ }
+
+ @Test
+ void splitResultsNumbersEveryRange() throws Exception {
+ Queue<Exchange> exchanges =
consumer("&splitResults=true").createExchanges(List.of(
+ range("A1:B2", List.of(List.of("a1", "b1"), List.of("a2",
"b2"))),
+ range("C1:D2", List.of(List.of("c1", "d1")))));
+
+ // one exchange per value, and the range index identifies which range
the value came from
+ assertThat(rangeIndexes(exchanges)).containsExactly(1, 1, 2);
+ assertThat(exchanges.stream().map(e ->
e.getIn().getHeader(GoogleSheetsStreamConstants.VALUE_INDEX)).toList())
+ .containsExactly(1, 2, 1);
+ }
+
+ @Test
+ void withoutSplitResultsEveryRangeIsOneExchange() throws Exception {
+ Queue<Exchange> exchanges = consumer("").createExchanges(List.of(
+ range("A1:B2", List.of(List.of("a1", "b1"))),
+ range("C1:D2", List.of(List.of("c1", "d1")))));
+
+ assertThat(rangeIndexes(exchanges)).containsExactly(1, 2);
+ }
+
+ @Test
+ void aRangeWithoutValuesIsNotAFailure() throws Exception {
+ // the sheets API omits the values field for a range that holds nothing
+ GoogleSheetsStreamConsumer consumer = consumer("&splitResults=true");
+ ValueRange empty = new
ValueRange().setRange("A1:B2").setMajorDimension("ROWS");
+
+ assertThatCode(() ->
consumer.createExchanges(List.of(empty))).doesNotThrowAnyException();
+ assertThat(consumer.createExchanges(List.of(empty))).isEmpty();
+ }
+
+ @Test
+ void aRangeWithoutValuesIsNotAFailureWhenResultsAreNotSplit() throws
Exception {
+ GoogleSheetsStreamConsumer consumer = consumer("&maxResults=1");
+ ValueRange empty = new
ValueRange().setRange("A1:B2").setMajorDimension("ROWS");
+
+ assertThatCode(() ->
consumer.createExchanges(List.of(empty))).doesNotThrowAnyException();
+ }
+}
