Google PubSub

https://cloud.google.com/pubsub/docs/spring

Spring 애플리케이션에서 Pub/Sub 사용  |  Google Cloud

 

https://github.com/GoogleCloudPlatform/spring-cloud-gcp/tree/main/spring-cloud-gcp-samples/spring-cloud-gcp-integration-pubsub-sample

GitHub - GoogleCloudPlatform/spring-cloud-gcp: New home for Spring Cloud GCP development starting with version 2.0.

 

1) auth json 파일 
 공식 가이드에는 환경변수를 잡는 것으로 되어 있던데 굳이 그럴 필요 없이 classpath에 위치시켜 주면 된다

 

application.yml

spring:
  cloud:
    gcp:
      credentials:
        location: classpath:/integrated-log.json
      project-id: integrated-log
      topic: jsonrpc-test-topic

integrated-log.json

{
	"type": "service_account",
	"project_id": "integrated-log",
	"private_key_id": "*********",
	"private_key": "*********",
	"client_email": "*********@*******.com",
	"client_id": "*********",
	"auth_uri": "https://accounts.google.com/o/oauth2/auth",
	"token_uri": "https://oauth2.googleapis.com/token",
	"auth_provider_x509_cert_url": "https://www.googleapis.com/oauth2/v1/certs",
	"client_x509_cert_url": "https://www.googleapis.com/robot/v1/metadata/x509/*********%40*********.com"
}

 

참조
https://velog.io/@caesars000/Spring-Boot%EC%97%90%EC%84%9C-Google-Cloud-Pub-%ED%99%9C%EC%9A%A9-%EC%98%88%EC%A0%9C

Spring Boot에서 Google Cloud Pub 활용 예제

 

 

2) 수신자(receiver)

 전반적인 내용은 예제와 같다

 

ReceiverConfig

@Configuration
public class ReceiverConfig 
{	
	Logger logger = LoggerFactory.getLogger(getClass());
	@Bean
	public MessageChannel inputMessageChannel()
	{
		return new PublishSubscribeChannel();
	}
	
	
	
	@Bean
	public PubSubInboundChannelAdapter inboundChannelAdapter(
		@Qualifier("inputMessageChannel") MessageChannel messageChannel,
		PubSubTemplate pubsubTemplate
	)
	{
		PubSubInboundChannelAdapter adapter = new PubSubInboundChannelAdapter(pubsubTemplate, "jsonrpc-test-subscription");
		adapter.setOutputChannel(messageChannel);
		adapter.setAckMode(AckMode.MANUAL);
		adapter.setPayloadType(String.class);
		return adapter;
	}
}

 

ReceiverComponent

@Component
public class ReceiverComponent 
{
	Logger logger = LoggerFactory.getLogger(this.getClass());
	@ServiceActivator(inputChannel = "inputMessageChannel")
	public void messageReceiver
	(
		String payload,
		@Header( GcpPubSubHeaders.ORIGINAL_MESSAGE) BasicAcknowledgeablePubsubMessage message
	)
	{
		logger.info("message arrived! payload: " + payload);
		message.ack();
	}

}

 

3) 발신자(Sender) 

 

SenderConfig

@Configuration
public class SenderConfig 
{
	Logger logger = LoggerFactory.getLogger(getClass());
	@Bean
	@ServiceActivator(inputChannel = "pubsubOutputChannel")
	public MessageHandler messageSender(PubSubTemplate pubsubTemplate)
	{
		PubSubMessageHandler adapter = new PubSubMessageHandler(pubsubTemplate, "jsonrpc-test-topic");
		 adapter.setSuccessCallback(
			      ((ackId, message) ->
			          logger.info("Message was sent via the outbound channel adapter to topic-two!")));

			  adapter.setFailureCallback(
			      (cause, message) -> logger.info("Error sending " + message + " due to " + cause));
		
		return adapter;
	}
}

 

SchedulerComponent

@Component
public class SchedulerComponent 
{

	Logger logger = LoggerFactory.getLogger(getClass());
	@Autowired
	PubSubOutboundGateway gateway;
	
	@Scheduled(fixedDelay = 60000)
	public void sendMessage()
	{
		logger.info("test");
		gateway.sendToPubSub("test");
	}
}