I'm trying to send push message using the emulator of pubsub, I'm using spring boot too, this is my configuration:
Dependency:
<dependency>
<groupId>org.springframework.cloud</groupId>
<artifactId>spring-cloud-gcp-starter-pubsub</artifactId>
</dependency>
My bean:
@Configuration
@AutoConfigureBefore(value= GcpPubSubAutoConfiguration.class)
@EnableConfigurationProperties(value= GcpPubSubProperties.class)
public class EmulatorPubSubConfiguration {
@Value("${spring.gcp.pubsub.projectid}")
private String projectId;
@Value("${spring.gcp.pubsub.subscriptorid}")
private String subscriptorId;
@Value("${spring.gcp.pubsub.topicid}")
private String topicId;
@Bean
public Publisher pubsubEmulator() throws IOException {
String hostport = System.getenv("PUBSUB_EMULATOR_HOST");
ManagedChannel channel = ManagedChannelBuilder.forTarget(hostport).usePlaintext().build();
try {
TransportChannelProvider channelProvider =
FixedTransportChannelProvider.create(GrpcTransportChannel.create(channel));
CredentialsProvider credentialsProvider = NoCredentialsProvider.create();
// Set the channel and credentials provider when creating a `TopicAdminClient`.
// Similarly for SubscriptionAdminClient
TopicAdminClient topicClient =
TopicAdminClient.create(
TopicAdminSettings.newBuilder()
.setTransportChannelProvider(channelProvider)
.setCredentialsProvider(credentialsProvider)
.build());
ProjectTopicName topicName = ProjectTopicName.of(projectId, topicId);
// Set the channel and credentials provider when creating a `Publisher`.
// Similarly for Subscriber
return Publisher.newBuilder(topicName)
.setChannelProvider(channelProvider)
.setCredentialsProvider(credentialsProvider)
.build();
} finally {
channel.shutdown();
}
}
}
Of course, I have set PUBSUB_EMULATOR_HOST system variable to localhost:8085, where is the emulator running
I created a rest controller for testing:
- for send push message
@Autowired
private Publisher pubsubPublisher;
@PostMapping("/send1")
public String publishMessage(@RequestParam("message") String message) throws InterruptedException, IOException {
Publisher pubsubPublisher = this.getPublisher();
ByteString data = ByteString.copyFromUtf8(message);
PubsubMessage pubsubMessage = PubsubMessage.newBuilder().setData(data).build();
ApiFuture<String> future = pubsubPublisher.publish(pubsubMessage);
//pubsubPublisher.publishAllOutstanding();
try {
// Add an asynchronous callback to handle success / failure
ApiFutures.addCallback(future,
new ApiFutureCallback<String>() {
@Override
public void onFailure(Throwable throwable) {
if (throwable instanceof ApiException) {
ApiException apiException = ((ApiException) throwable);
// details on the API exception
System.out.println(apiException.getStatusCode().getCode());
System.out.println(apiException.isRetryable());
}
System.out.println("Error publishing message : " + message);
System.out.println("Error publishing error : " + throwable.getMessage());
System.out.println("Error publishing cause : " + throwable.getCause());
}
@Override
public void onSuccess(String messageId) {
// Once published, returns server-assigned message ids (unique within the topic)
System.out.println(messageId);
}
},
MoreExecutors.directExecutor());
}
finally {
if (pubsubPublisher != null) {
// When finished with the publisher, shutdown to free up resources.
pubsubPublisher.shutdown();
pubsubPublisher.awaitTermination(1, TimeUnit.MINUTES);
}
}
return "ok";
- for get message:
@PostMapping("/pushtest")
public String pushTest(@RequestBody CloudPubSubPushMessage request) {
System.out.println( "------> message received: " + decode(request.getMessage().getData()) );
return request.toString();
}
I have created my topic and subscription in the emulator, I followed this tutorial:
https://cloud.google.com/pubsub/docs/emulator
I'm set the endpoint "pushtest" for get push message in the emulator, with this command:
python subscriber.py PUBSUB_PROJECT_ID create-push TOPIC_ID SUBSCRIPTION_ID PUSH_ENDPOINT
But when I run the test, doesn't reach "/pushtest" endpoint and I'm getting this error:
Task java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask@265d5d05 [Not completed, task = java.util.concurrent.Executors$RunnableAdapter@a8c8be3 [Wrapped task = com.google.common.util.concurrent.TrustedListenableFutureTask@1a53c57c [status=PENDING, info=[task=[running=[NOT STARTED YET], com.google.api.gax.rpc.AttemptCallable@3866e1d0]]]]] rejected from java.util.concurrent.ScheduledThreadPoolExecutor@3f34809a [Terminated, pool size = 0, active threads = 0, queued tasks = 0, completed tasks = 1]
for assurance that the emulator is running ok, I'm run the test in python with the following command:
python publisher.py PUBSUB_PROJECT_ID publish TOPIC_ID
And I'm getting messages correctly in "pushtest" endpoint.
I don't know why sorry for my hazing.
Thanks for your help.
