This is an automated email from the ASF dual-hosted git repository.
jamesnetherton pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel-quarkus.git
The following commit(s) were added to refs/heads/main by this push:
new 6b96370ad0 Document and test Avro trusted classes for the Salesforce
Pub / Sub API
6b96370ad0 is described below
commit 6b96370ad07dc1e4b4a30854fc5b23501f6a9cf5
Author: James Netherton <[email protected]>
AuthorDate: Mon Oct 12 19:57:32 2026 +0100
Document and test Avro trusted classes for the Salesforce Pub / Sub API
* Correct the AvroGenerated claim in the 3.40.0 migration guide
The guide said that Avro added @AvroGenerated in 1.9.0. The annotation and
its
use in generated records date back to at least Avro 1.7.4, so specific
records
generated before 1.9.0 are trusted automatically like any other.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
* Document and test Avro trusted classes for the Salesforce Pub / Sub API
Avro 1.12.2 only maps events to classes that are trusted. A subscription
with
pubSubDeserializeType=POJO therefore fails with a SecurityException unless
the
pubSubPojoClass is trusted via quarkus.avro.trusted-classes or
quarkus.avro.trusted-packages. The class is an endpoint option, so the
extension cannot determine it at build time.
Document what needs to be trusted for each pubSubDeserializeType in the
Salesforce extension docs and in the 3.40.0 migration guide.
The Pub / Sub API tests only ran with Salesforce credentials, so the failure
went unnoticed. Test subscribing with each pubSubDeserializeType against a
mock
Pub / Sub API gRPC server when no credentials are provided.
Co-Authored-By: Claude Opus 5.5 <[email protected]>
---------
Co-authored-by: Claude Opus 5.5 <[email protected]>
---
.../modules/ROOT/pages/migration-guide/3.40.0.adoc | 26 +++-
.../pages/reference/extensions/salesforce.adoc | 52 ++++++++
.../salesforce/runtime/src/main/doc/usage.adoc | 51 ++++++++
integration-tests/salesforce/README.adoc | 2 +
.../component/salesforce/SalesforceResource.java | 8 ++
.../component/salesforce/SalesforceRoutes.java | 60 ++++-----
.../src/main/resources/application.properties | 2 +
.../component/salesforce/MockPubSubService.java | 134 +++++++++++++++++++++
.../component/salesforce/SalesforcePubSubIT.java | 26 ++++
.../component/salesforce/SalesforcePubSubTest.java | 88 ++++++++++++++
.../salesforce/SalesforceTestResource.java | 40 +++++-
11 files changed, 458 insertions(+), 31 deletions(-)
diff --git a/docs/modules/ROOT/pages/migration-guide/3.40.0.adoc
b/docs/modules/ROOT/pages/migration-guide/3.40.0.adoc
index 49d5fc0a8c..33daeb4aaf 100644
--- a/docs/modules/ROOT/pages/migration-guide/3.40.0.adoc
+++ b/docs/modules/ROOT/pages/migration-guide/3.40.0.adoc
@@ -265,7 +265,7 @@ java.lang.SecurityException: Forbidden
com.acme.orders.avro.Order! This class is
Classes annotated with `@org.apache.avro.specific.AvroGenerated` are trusted
automatically, which covers everything produced by the `generate-code` goal of
`quarkus-maven-plugin` from `src/main/avro`. Nothing needs to be done for those.
-Specific records that predate that annotation, which Avro added in 1.9.0, or
that were written by hand, have to be trusted explicitly. Prefer the Quarkus
configuration properties, which are resolved at build time and so apply in both
JVM and native mode:
+Specific records without that annotation, such as those written by hand, have
to be trusted explicitly. Prefer the Quarkus configuration properties, which
are resolved at build time and so apply in both JVM and native mode:
[source,properties]
----
@@ -276,3 +276,27 @@ quarkus.avro.trusted-classes=com.acme.LegacyOrder
Avro's own `org.apache.avro.SERIALIZABLE_PACKAGES` and
`org.apache.avro.SERIALIZABLE_CLASSES` system properties still work as an
addendum, since Quarkus adds its trusted classes on top of them rather than
replacing them. They are read once, when Avro initializes, so they have to be
set on the command line. Note that `*` in
`org.apache.avro.SERIALIZABLE_PACKAGES` disables the class loading protection
entirely.
Whichever route is used, list only the packages holding your own Avro model
classes.
+
+== Salesforce extension changes
+
+=== Pub / Sub API event classes must be trusted by Avro
+
+The Salesforce Pub / Sub API encodes events with Avro, so the Avro change
described above also applies to routes that use `pubSubSubscribe` or
`pubSubPublish`.
+
+The most likely case to be affected is a subscription with
`pubSubDeserializeType=POJO`. The class configured on `pubSubPojoClass` is a
plain class, so it is never trusted automatically and each event now fails with:
+
+[source,text]
+----
+java.lang.SecurityException: Forbidden org.foo.TestEvent! This class is not
trusted to be included in Avro schemas.
+----
+
+Trust the class, or its package, in `application.properties`.
+
+[source,properties]
+----
+quarkus.avro.trusted-classes=org.foo.TestEvent
+----
+
+Subscriptions with the default `AVRO` type, or with `SPECIFIC_RECORD`, map
events to the Avro `SpecificRecord` classes in the `com.sforce.eventbus`
package. Those are trusted automatically if they are annotated with
`@org.apache.avro.specific.AvroGenerated`, and only need to be trusted
explicitly if they lack that annotation, for example because they were written
by hand. The `GENERIC_RECORD` and `JSON` types are unaffected.
+
+Refer to the
xref:reference/extensions/salesforce.adoc#extensions-salesforce-usage-trusting-classes-used-with-the-pub-sub-api[Salesforce
extension documentation] for details.
diff --git a/docs/modules/ROOT/pages/reference/extensions/salesforce.adoc
b/docs/modules/ROOT/pages/reference/extensions/salesforce.adoc
index b386fe2393..9951815e60 100644
--- a/docs/modules/ROOT/pages/reference/extensions/salesforce.adoc
+++ b/docs/modules/ROOT/pages/reference/extensions/salesforce.adoc
@@ -84,6 +84,58 @@ To generate Salesforce DTOs for your project, use the
`salesforce-maven-plugin`.
</plugin>
----
+[id="extensions-salesforce-usage-trusting-classes-used-with-the-pub-sub-api"]
+=== Trusting classes used with the Pub / Sub API
+
+The Pub / Sub API exchanges events encoded with Avro. Since Avro 1.12.2,
classes that events are mapped to must be explicitly trusted, otherwise the
event is rejected with:
+
+[source,text]
+----
+java.lang.SecurityException: Forbidden org.foo.TestEvent! This class is not
trusted to be included in Avro schemas.
+----
+
+Whether anything needs to be trusted depends on the `pubSubDeserializeType` of
the subscribing endpoint.
+
+[cols="1,3"]
+|===
+|`pubSubDeserializeType` |Classes to trust
+
+|`AVRO` (default), `SPECIFIC_RECORD`
+|The Avro `SpecificRecord` classes in the `com.sforce.eventbus` package that
events are mapped to. Classes annotated with
`@org.apache.avro.specific.AvroGenerated` are trusted automatically, which
covers classes generated by the Avro compiler. Only classes without that
annotation, such as those written by hand, need to be trusted explicitly.
+
+|`POJO`
+|The class configured on the `pubSubPojoClass` option. This is always
required, as it is a plain class that Avro maps to reflectively.
+
+|`GENERIC_RECORD`, `JSON`
+|None.
+|===
+
+For example, given the following route.
+
+[source,java]
+----
+from("salesforce:pubSubSubscribe:/event/TestEvent__e?pubSubDeserializeType=POJO&pubSubPojoClass=org.foo.TestEvent")
+ .log("Received Salesforce POJO topic message: ${body}");
+----
+
+Class `org.foo.TestEvent` would need to be trusted in `application.properties`.
+
+[source,properties]
+----
+quarkus.avro.trusted-classes=org.foo.TestEvent
+----
+
+Alternatively, trust every class in a package and its sub-packages. List only
the packages holding your own event classes.
+
+[source,properties]
+----
+quarkus.avro.trusted-packages=org.foo
+----
+
+Publishing with `pubSubPublish` does not require the message body class to be
trusted. However, if an Avro `SpecificRecord` class exists for the topic in the
`com.sforce.eventbus` package, then publishing a POJO body requires that class
to be trusted in the same way as for subscribing.
+
+Refer to the
xref:reference/extensions/avro.adoc#extensions-avro-configuration-trusted-classes[Avro
extension documentation] for more information.
+
[id="extensions-salesforce-usage-native-mode-support-for-pub-sub-api-with-pojo-pubsubdeserializetype"]
=== Native mode support for Pub / Sub API with POJO `pubSubDeserializeType`
diff --git a/extensions/salesforce/runtime/src/main/doc/usage.adoc
b/extensions/salesforce/runtime/src/main/doc/usage.adoc
index b6b9becd5b..ca4b0e9589 100644
--- a/extensions/salesforce/runtime/src/main/doc/usage.adoc
+++ b/extensions/salesforce/runtime/src/main/doc/usage.adoc
@@ -34,6 +34,57 @@ To generate Salesforce DTOs for your project, use the
`salesforce-maven-plugin`.
</plugin>
----
+=== Trusting classes used with the Pub / Sub API
+
+The Pub / Sub API exchanges events encoded with Avro. Since Avro 1.12.2,
classes that events are mapped to must be explicitly trusted, otherwise the
event is rejected with:
+
+[source,text]
+----
+java.lang.SecurityException: Forbidden org.foo.TestEvent! This class is not
trusted to be included in Avro schemas.
+----
+
+Whether anything needs to be trusted depends on the `pubSubDeserializeType` of
the subscribing endpoint.
+
+[cols="1,3"]
+|===
+|`pubSubDeserializeType` |Classes to trust
+
+|`AVRO` (default), `SPECIFIC_RECORD`
+|The Avro `SpecificRecord` classes in the `com.sforce.eventbus` package that
events are mapped to. Classes annotated with
`@org.apache.avro.specific.AvroGenerated` are trusted automatically, which
covers classes generated by the Avro compiler. Only classes without that
annotation, such as those written by hand, need to be trusted explicitly.
+
+|`POJO`
+|The class configured on the `pubSubPojoClass` option. This is always
required, as it is a plain class that Avro maps to reflectively.
+
+|`GENERIC_RECORD`, `JSON`
+|None.
+|===
+
+For example, given the following route.
+
+[source,java]
+----
+from("salesforce:pubSubSubscribe:/event/TestEvent__e?pubSubDeserializeType=POJO&pubSubPojoClass=org.foo.TestEvent")
+ .log("Received Salesforce POJO topic message: ${body}");
+----
+
+Class `org.foo.TestEvent` would need to be trusted in `application.properties`.
+
+[source,properties]
+----
+quarkus.avro.trusted-classes=org.foo.TestEvent
+----
+
+Alternatively, trust every class in a package and its sub-packages. List only
the packages holding your own event classes.
+
+[source,properties]
+----
+quarkus.avro.trusted-packages=org.foo
+----
+
+Publishing with `pubSubPublish` does not require the message body class to be
trusted. However, if an Avro `SpecificRecord` class exists for the topic in the
`com.sforce.eventbus` package, then publishing a POJO body requires that class
to be trusted in the same way as for subscribing.
+
+Refer to the
xref:reference/extensions/avro.adoc#extensions-avro-configuration-trusted-classes[Avro
extension documentation] for more information.
+
=== Native mode support for Pub / Sub API with POJO `pubSubDeserializeType`
When using the Camel Salesforce Pub / Sub API and `pubSubDeserializeType` is
configured as `POJO`, you must register any classes configured on the
`pubSubPojoClass` option for reflection.
diff --git a/integration-tests/salesforce/README.adoc
b/integration-tests/salesforce/README.adoc
index df31dfe5fc..362adffc17 100644
--- a/integration-tests/salesforce/README.adoc
+++ b/integration-tests/salesforce/README.adoc
@@ -4,6 +4,8 @@ By default, the Salesforce integration tests use WireMock to
stub the API intera
IMPORTANT: Note that when using Wiremock, the Change Data Capture Events test,
and the Platform event test are disabled.
+The Pub / Sub API uses gRPC, which WireMock cannot stub. A mock Pub / Sub API
server is started alongside WireMock instead, and `SalesforcePubSubTest` tests
subscribing to events with each `pubSubDeserializeType`. Publishing events is
not covered, since the Pub / Sub API client can only publish over TLS. When
Salesforce credentials are provided, the mock server is not started,
`SalesforcePubSubTest` is disabled, and the Pub / Sub API is tested against
Salesforce by `SalesforceIntegrationTest`.
+
To run `camel-quarkus-salesforce` integration tests using Salesforce API
interactions, you must first create a Salesforce developer account
https://developer.salesforce.com/.
Next create a new 'Connected App' by visiting Apps -> App Manager page. In
the form, enable OAuth Settings, and set those fields.
diff --git
a/integration-tests/salesforce/src/main/java/org/apache/camel/quarkus/component/salesforce/SalesforceResource.java
b/integration-tests/salesforce/src/main/java/org/apache/camel/quarkus/component/salesforce/SalesforceResource.java
index 9b35ca899e..f2624e56c6 100644
---
a/integration-tests/salesforce/src/main/java/org/apache/camel/quarkus/component/salesforce/SalesforceResource.java
+++
b/integration-tests/salesforce/src/main/java/org/apache/camel/quarkus/component/salesforce/SalesforceResource.java
@@ -46,6 +46,7 @@ import org.apache.camel.CamelContext;
import org.apache.camel.ConsumerTemplate;
import org.apache.camel.Exchange;
import org.apache.camel.FluentProducerTemplate;
+import org.apache.camel.component.salesforce.PubSubApiConsumer;
import org.apache.camel.component.salesforce.SalesforceEndpointConfig;
import org.apache.camel.component.salesforce.api.dto.Attributes;
import org.apache.camel.component.salesforce.api.dto.CreateSObjectResult;
@@ -66,6 +67,7 @@ import
org.apache.camel.quarkus.component.salesforce.generated.QueryRecordsAccou
import
org.apache.camel.quarkus.component.salesforce.model.GlobalObjectsAndHeaders;
import org.apache.camel.quarkus.component.salesforce.model.TestEventPojo;
import org.apache.camel.spi.RouteController;
+import org.eclipse.microprofile.config.ConfigProvider;
@Path("/salesforce")
public class SalesforceResource {
@@ -172,6 +174,12 @@ public class SalesforceResource {
throws Exception {
RouteController controller = context.getRouteController();
if (action.equals("start")) {
+ // Set by the test resource when the mock Pub / Sub API server is
used, as it does not support TLS
+ if (context.getRoute(routeId).getConsumer() instanceof
PubSubApiConsumer consumer
+ &&
ConfigProvider.getConfig().getOptionalValue("salesforce.pubsub.use-plain-text",
Boolean.class)
+ .orElse(false)) {
+ consumer.setUsePlainTextConnection(true);
+ }
controller.startRoute(routeId);
} else if (action.equals("stop")) {
controller.stopRoute(routeId);
diff --git
a/integration-tests/salesforce/src/main/java/org/apache/camel/quarkus/component/salesforce/SalesforceRoutes.java
b/integration-tests/salesforce/src/main/java/org/apache/camel/quarkus/component/salesforce/SalesforceRoutes.java
index 60ea50c2b7..cd31488339 100644
---
a/integration-tests/salesforce/src/main/java/org/apache/camel/quarkus/component/salesforce/SalesforceRoutes.java
+++
b/integration-tests/salesforce/src/main/java/org/apache/camel/quarkus/component/salesforce/SalesforceRoutes.java
@@ -102,35 +102,37 @@ public class SalesforceRoutes extends RouteBuilder {
from("direct:pubSubPublish")
.log("Sent Salesforce topic message: ${body}")
.to("salesforce:pubSubPublish:/event/TestEvent__e");
-
- // AVRO deserialize type (also analogous to SPECIFIC_RECORD)
- from("salesforce:pubSubSubscribe:/event/TestEvent__e")
- .routeId("topicSubscribeAvro")
- .autoStartup(false)
- .log("Received Salesforce AVRO topic message: ${body}")
- .to("seda:pubSubSubscribeAvro");
-
- // GENERIC_RECORD deserialize type
-
from("salesforce:pubSubSubscribe:/event/TestEvent__e?pubSubDeserializeType=GENERIC_RECORD")
- .routeId("topicSubscribeGenericRecord")
- .autoStartup(false)
- .log("Received Salesforce GENERIC_RECORD topic message:
${body}")
- .to("seda:pubSubSubscribeGenericRecord");
-
- // JSON deserialize type
-
from("salesforce:pubSubSubscribe:/event/TestEvent__e?pubSubDeserializeType=JSON")
- .routeId("topicSubscribeJson")
- .autoStartup(false)
- .log("Received Salesforce JSON topic message: ${body}")
- .to("seda:pubSubSubscribeJson");
-
- // POJO deserialize type
-
from("salesforce:pubSubSubscribe:/event/TestEvent__e?pubSubDeserializeType=POJO&pubSubPojoClass="
- + TestEventPojo.class.getName())
- .routeId("topicSubscribePojo")
- .autoStartup(false)
- .log("Received Salesforce POJO topic message: ${body}")
- .to("seda:pubSubSubscribePojo");
}
+
+ // The Pub / Sub API subscribers work with real credentials, or a mock
Pub / Sub API server
+
+ // AVRO deserialize type (also analogous to SPECIFIC_RECORD)
+ from("salesforce:pubSubSubscribe:/event/TestEvent__e")
+ .routeId("topicSubscribeAvro")
+ .autoStartup(false)
+ .log("Received Salesforce AVRO topic message: ${body}")
+ .to("seda:pubSubSubscribeAvro");
+
+ // GENERIC_RECORD deserialize type
+
from("salesforce:pubSubSubscribe:/event/TestEvent__e?pubSubDeserializeType=GENERIC_RECORD")
+ .routeId("topicSubscribeGenericRecord")
+ .autoStartup(false)
+ .log("Received Salesforce GENERIC_RECORD topic message:
${body}")
+ .to("seda:pubSubSubscribeGenericRecord");
+
+ // JSON deserialize type
+
from("salesforce:pubSubSubscribe:/event/TestEvent__e?pubSubDeserializeType=JSON")
+ .routeId("topicSubscribeJson")
+ .autoStartup(false)
+ .log("Received Salesforce JSON topic message: ${body}")
+ .to("seda:pubSubSubscribeJson");
+
+ // POJO deserialize type
+
from("salesforce:pubSubSubscribe:/event/TestEvent__e?pubSubDeserializeType=POJO&pubSubPojoClass="
+ + TestEventPojo.class.getName())
+ .routeId("topicSubscribePojo")
+ .autoStartup(false)
+ .log("Received Salesforce POJO topic message: ${body}")
+ .to("seda:pubSubSubscribePojo");
}
}
diff --git
a/integration-tests/salesforce/src/main/resources/application.properties
b/integration-tests/salesforce/src/main/resources/application.properties
index f85470aa24..b01a803795 100644
--- a/integration-tests/salesforce/src/main/resources/application.properties
+++ b/integration-tests/salesforce/src/main/resources/application.properties
@@ -20,3 +20,5 @@
#
camel.component.salesforce.loginUrl = https://login.salesforce.com
+# The class used with pubSubDeserializeType=POJO must be trusted by Avro
+quarkus.avro.trusted-classes =
org.apache.camel.quarkus.component.salesforce.model.TestEventPojo
diff --git
a/integration-tests/salesforce/src/test/java/org/apache/camel/quarkus/component/salesforce/MockPubSubService.java
b/integration-tests/salesforce/src/test/java/org/apache/camel/quarkus/component/salesforce/MockPubSubService.java
new file mode 100644
index 0000000000..d6e0885b09
--- /dev/null
+++
b/integration-tests/salesforce/src/test/java/org/apache/camel/quarkus/component/salesforce/MockPubSubService.java
@@ -0,0 +1,134 @@
+/*
+ * 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.quarkus.component.salesforce;
+
+import java.io.ByteArrayOutputStream;
+import java.io.IOException;
+
+import com.google.protobuf.ByteString;
+import com.salesforce.eventbus.protobuf.ConsumerEvent;
+import com.salesforce.eventbus.protobuf.FetchRequest;
+import com.salesforce.eventbus.protobuf.FetchResponse;
+import com.salesforce.eventbus.protobuf.ProducerEvent;
+import com.salesforce.eventbus.protobuf.PubSubGrpc;
+import com.salesforce.eventbus.protobuf.SchemaInfo;
+import com.salesforce.eventbus.protobuf.SchemaRequest;
+import com.salesforce.eventbus.protobuf.TopicInfo;
+import com.salesforce.eventbus.protobuf.TopicRequest;
+import io.grpc.stub.StreamObserver;
+import org.apache.avro.Schema;
+import org.apache.avro.generic.GenericDatumWriter;
+import org.apache.avro.generic.GenericRecord;
+import org.apache.avro.generic.GenericRecordBuilder;
+import org.apache.avro.io.BinaryEncoder;
+import org.apache.avro.io.EncoderFactory;
+
+/**
+ * A minimal stand-in for the Salesforce Pub / Sub API, which delivers a
single TestEvent__e event to each subscriber.
+ */
+public class MockPubSubService extends PubSubGrpc.PubSubImplBase {
+ public static final String CREATED_BY = "0055g00000AbCdEfGH";
+ public static final long CREATED_DATE = 1760000000000L;
+ public static final String TEST_FIELD_VALUE = "Camel Quarkus Pub / Sub
test";
+
+ private static final String SCHEMA_ID = "test-event-schema";
+ // Matches the schema of the generated com.sforce.eventbus.TestEvent__e
+ private static final String SCHEMA_JSON = """
+ {
+ "type": "record",
+ "name": "TestEvent__e",
+ "namespace": "com.sforce.eventbus",
+ "fields": [
+ {"name": "CreatedDate", "type": "long", "doc":
"CreatedDate:DateTime"},
+ {"name": "CreatedById", "type": "string", "doc":
"CreatedBy:EntityId"},
+ {"name": "Test_Field__c", "type": ["null", "string"], "doc":
"Data:Text", "default": null}
+ ]
+ }
+ """;
+
+ @Override
+ public void getTopic(TopicRequest request, StreamObserver<TopicInfo>
responseObserver) {
+ responseObserver.onNext(TopicInfo.newBuilder()
+ .setTopicName(request.getTopicName())
+ .setCanPublish(true)
+ .setCanSubscribe(true)
+ .setSchemaId(SCHEMA_ID)
+ .build());
+ responseObserver.onCompleted();
+ }
+
+ @Override
+ public void getSchema(SchemaRequest request, StreamObserver<SchemaInfo>
responseObserver) {
+ responseObserver.onNext(SchemaInfo.newBuilder()
+ .setSchemaId(SCHEMA_ID)
+ .setSchemaJson(SCHEMA_JSON)
+ .build());
+ responseObserver.onCompleted();
+ }
+
+ @Override
+ public StreamObserver<FetchRequest>
subscribe(StreamObserver<FetchResponse> responseObserver) {
+ return new StreamObserver<>() {
+ @Override
+ public void onNext(FetchRequest request) {
+ ByteString replayId = ByteString.copyFromUtf8("1");
+ ConsumerEvent event = ConsumerEvent.newBuilder()
+ .setReplayId(replayId)
+ .setEvent(ProducerEvent.newBuilder()
+ .setId("test-event")
+ .setSchemaId(SCHEMA_ID)
+
.setPayload(ByteString.copyFrom(createPayload())))
+ .build();
+
+ // Report the remaining requested events as pending, so that
the client does not ask for more
+ responseObserver.onNext(FetchResponse.newBuilder()
+ .addEvents(event)
+ .setLatestReplayId(replayId)
+ .setPendingNumRequested(request.getNumRequested())
+ .build());
+ }
+
+ @Override
+ public void onError(Throwable t) {
+ // Noop
+ }
+
+ @Override
+ public void onCompleted() {
+ responseObserver.onCompleted();
+ }
+ };
+ }
+
+ private static byte[] createPayload() {
+ Schema schema = new Schema.Parser().parse(SCHEMA_JSON);
+ GenericRecord record = new GenericRecordBuilder(schema)
+ .set("CreatedDate", CREATED_DATE)
+ .set("CreatedById", CREATED_BY)
+ .set("Test_Field__c", TEST_FIELD_VALUE)
+ .build();
+
+ try {
+ ByteArrayOutputStream out = new ByteArrayOutputStream();
+ BinaryEncoder encoder =
EncoderFactory.get().directBinaryEncoder(out, null);
+ new GenericDatumWriter<GenericRecord>(schema).write(record,
encoder);
+ return out.toByteArray();
+ } catch (IOException e) {
+ throw new IllegalStateException(e);
+ }
+ }
+}
diff --git
a/integration-tests/salesforce/src/test/java/org/apache/camel/quarkus/component/salesforce/SalesforcePubSubIT.java
b/integration-tests/salesforce/src/test/java/org/apache/camel/quarkus/component/salesforce/SalesforcePubSubIT.java
new file mode 100644
index 0000000000..deeb5aee55
--- /dev/null
+++
b/integration-tests/salesforce/src/test/java/org/apache/camel/quarkus/component/salesforce/SalesforcePubSubIT.java
@@ -0,0 +1,26 @@
+/*
+ * 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.quarkus.component.salesforce;
+
+import io.quarkus.test.junit.QuarkusIntegrationTest;
+import org.junit.jupiter.api.condition.EnabledIf;
+
+@EnabledIf("org.apache.camel.quarkus.component.salesforce.SalesforceTestResource#isMockBackend")
+@QuarkusIntegrationTest
+class SalesforcePubSubIT extends SalesforcePubSubTest {
+
+}
diff --git
a/integration-tests/salesforce/src/test/java/org/apache/camel/quarkus/component/salesforce/SalesforcePubSubTest.java
b/integration-tests/salesforce/src/test/java/org/apache/camel/quarkus/component/salesforce/SalesforcePubSubTest.java
new file mode 100644
index 0000000000..e824b7b499
--- /dev/null
+++
b/integration-tests/salesforce/src/test/java/org/apache/camel/quarkus/component/salesforce/SalesforcePubSubTest.java
@@ -0,0 +1,88 @@
+/*
+ * 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.quarkus.component.salesforce;
+
+import io.quarkus.test.common.QuarkusTestResource;
+import io.quarkus.test.junit.QuarkusTest;
+import io.restassured.RestAssured;
+import io.restassured.response.ValidatableResponse;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.condition.EnabledIf;
+
+import static org.hamcrest.Matchers.is;
+
+/**
+ * Tests the Pub / Sub API subscriber against {@link MockPubSubService}. When
Salesforce credentials are provided,
+ * the Pub / Sub API is tested against Salesforce in {@link
SalesforceIntegrationTest} instead.
+ */
+@EnabledIf("org.apache.camel.quarkus.component.salesforce.SalesforceTestResource#isMockBackend")
+@QuarkusTest
+@QuarkusTestResource(SalesforceTestResource.class)
+class SalesforcePubSubTest {
+
+ @Test
+ void testPubSubAvro() {
+ subscribe("topicSubscribeAvro", "/salesforce/subscribe/event/avro")
+ .body(
+ "createdBy", is(MockPubSubService.CREATED_BY),
+ "createdDate", is(MockPubSubService.CREATED_DATE),
+ "testFieldValue",
is(MockPubSubService.TEST_FIELD_VALUE));
+ }
+
+ @Test
+ void testPubSubGenericRecord() {
+ subscribe("topicSubscribeGenericRecord",
"/salesforce/subscribe/event/generic/record")
+ .body(
+ "createdBy", is(MockPubSubService.CREATED_BY),
+ "createdDate",
is(String.valueOf(MockPubSubService.CREATED_DATE)),
+ "testFieldValue",
is(MockPubSubService.TEST_FIELD_VALUE));
+ }
+
+ @Test
+ void testPubSubJson() {
+ subscribe("topicSubscribeJson", "/salesforce/subscribe/event/json")
+ .body(
+ "CreatedById", is(MockPubSubService.CREATED_BY),
+ "CreatedDate", is(MockPubSubService.CREATED_DATE),
+ "Test_Field__c",
is(MockPubSubService.TEST_FIELD_VALUE));
+ }
+
+ @Test
+ void testPubSubPojo() {
+ subscribe("topicSubscribePojo", "/salesforce/subscribe/event/pojo")
+ .body(
+ "createdBy", is(MockPubSubService.CREATED_BY),
+ "createdDate", is(MockPubSubService.CREATED_DATE),
+ "testFieldValue",
is(MockPubSubService.TEST_FIELD_VALUE));
+ }
+
+ private static ValidatableResponse subscribe(String routeId, String path) {
+ try {
+ RestAssured.post("/salesforce/route/" + routeId + "/start")
+ .then()
+ .statusCode(200);
+
+ return RestAssured.get(path)
+ .then()
+ .statusCode(200);
+ } finally {
+ RestAssured.post("/salesforce/route/" + routeId + "/stop")
+ .then()
+ .statusCode(200);
+ }
+ }
+}
diff --git
a/integration-tests/salesforce/src/test/java/org/apache/camel/quarkus/component/salesforce/SalesforceTestResource.java
b/integration-tests/salesforce/src/test/java/org/apache/camel/quarkus/component/salesforce/SalesforceTestResource.java
index afa81b420e..4e40613c82 100644
---
a/integration-tests/salesforce/src/test/java/org/apache/camel/quarkus/component/salesforce/SalesforceTestResource.java
+++
b/integration-tests/salesforce/src/test/java/org/apache/camel/quarkus/component/salesforce/SalesforceTestResource.java
@@ -16,8 +16,12 @@
*/
package org.apache.camel.quarkus.component.salesforce;
+import java.io.IOException;
import java.util.Map;
+import java.util.stream.Stream;
+import io.grpc.Server;
+import io.grpc.ServerBuilder;
import
org.apache.camel.quarkus.test.wiremock.WireMockTestResourceLifecycleManager;
public class SalesforceTestResource extends
WireMockTestResourceLifecycleManager {
@@ -27,6 +31,16 @@ public class SalesforceTestResource extends
WireMockTestResourceLifecycleManager
private static final String SALESFORCE_USERNAME = "SALESFORCE_USERNAME";
private static final String SALESFORCE_PASSWORD = "SALESFORCE_PASSWORD";
+ private Server pubSubServer;
+
+ /**
+ * Whether the tests run against mocked Salesforce APIs, which is the case
unless Salesforce credentials are provided
+ */
+ public static boolean isMockBackend() {
+ return Stream.of(SALESFORCE_CLIENT_ID, SALESFORCE_CLIENT_SECRET,
SALESFORCE_USERNAME, SALESFORCE_PASSWORD)
+ .anyMatch(envVar -> System.getenv(envVar) == null);
+ }
+
@Override
protected String getRecordTargetBaseUrl() {
return SALESFORCE_BASE_URL;
@@ -34,7 +48,7 @@ public class SalesforceTestResource extends
WireMockTestResourceLifecycleManager
@Override
protected boolean isMockingEnabled() {
- return !envVarsPresent(SALESFORCE_CLIENT_ID, SALESFORCE_CLIENT_SECRET,
SALESFORCE_USERNAME, SALESFORCE_PASSWORD);
+ return isMockBackend();
}
@Override
@@ -46,6 +60,30 @@ public class SalesforceTestResource extends
WireMockTestResourceLifecycleManager
options.put("camel.component.salesforce.password", "password");
options.put("camel.component.salesforce.clientId", "clientId");
}
+
+ // The Pub / Sub API uses gRPC, which WireMock cannot stub
+ if (isMockingEnabled()) {
+ try {
+ pubSubServer = ServerBuilder.forPort(0)
+ .addService(new MockPubSubService())
+ .build()
+ .start();
+ } catch (IOException e) {
+ throw new RuntimeException("Failed to start the mock Pub / Sub
API server", e);
+ }
+ options.put("camel.component.salesforce.pubSubHost", "localhost");
+ options.put("camel.component.salesforce.pubSubPort",
String.valueOf(pubSubServer.getPort()));
+ // The mock Pub / Sub API server does not support TLS
+ options.put("salesforce.pubsub.use-plain-text", "true");
+ }
return options;
}
+
+ @Override
+ public void stop() {
+ if (pubSubServer != null) {
+ pubSubServer.shutdownNow();
+ }
+ super.stop();
+ }
}
