Spring Cloud StreamにProtocol Buffersを対応させる


![Kobito.D5Bp80.png](https://qiita-image-store.s3.amazonaws.com/0/28298/716f3894-91e6-b552-56dd-a941d79e39bf.png "Kobito.D5Bp80.png")


## なぜSpring Cloud StreamとProtocol Buffersなのか

すごく簡単に言えば

### _メッセージングを楽にしたい_

からです。

最近のシステムは同期的なAPIコールの他に、スケーラビリティなどを考慮した非同期メッセージングを用いるケースも増えてきました。

ここで問題になるのが、複数システムを跨ぐため、そのスキーマをどのように管理するかです。

この場合REST API定義のように、共有可能で、シリアライズ形式に後方互換性あるものが望ましいです。

メッセージングによく使われるJSONのスキーマ定義としてメジャーなのがJSON Schemaですが、その定義は決して[扱いやすいとはいいにくい](https://qiita.com/yugui/items/160737021d25d761b353#protobuf)ものです。

そこで候補に上がるのがgRPCにも採用されている[Protocol Buffers](https://developers.google.com/protocol-buffers/)です。

Protocol BuffersはIDL(インタフェース定義言語)の為、JSON Schemaの用に、従来の仕様に縛られず、簡潔にそのスキーマを定義することができます。

ということで、Javaでよく使われるSpringのメッセージングのフレームワークである[Spring Cloud Stream](https://cloud.spring.io/spring-cloud-stream/)上で、Protocol Buffersによるメッセージングを行ってみます。

## 基本的な仕組み

Spring Cloud Streamではデフォルトの変換フォーマットとして、Jsonがサポートされていますが、他のフォーマットにも対応するために`MessageConverter`クラスが用意されいてます。

なので、やることは単純で
1. Protocol BuffersのMessageConverter実装を作成
2. 使用するアプリケーション内でBean登録
3. application.yaml(propeties)のstreamのcontent-typeにProtocol BuffersのMIMEを指定
の3点です。

## 実装方法

順に具体的な実装を行っていきます。

### 1. Protocol BuffersのMessageConverter実装を作成
サンプルなので細かなエラー処理などは省いていますが、以下のようなMessageConverterの実装クラスを用意します。

```Java
public class ProtobufMessageConverter extends AbstractMessageConverter {

	public ProtobufMessageConverter() {
	    // Protocol Buffers用のMIMEを用意
		super(new MimeType("application", "x-protobuf"));
	}

	@Override
	protected boolean supports(Class<?> clazz) {
	    // Protocol Buffersのメッセージに使用されるスーパークラスをサポートクラスに指定
		return AbstractMessageLite.class.isAssignableFrom(clazz);
	}

	@Override
	protected Object convertToInternal(Object payload, MessageHeaders headers, Object conversionHint) {
	    // Protocol Buffersの機能でバイト配列への変換
		return ((AbstractMessageLite) payload).toByteArray();
	}

	@Override
	@lombok.SneakyThrows
	protected Object convertFromInternal(Message<?> message, Class<?> targetClass, Object conversionHint) {
		Object payload = message.getPayload();

		// conversionHintのMethodParameterから、受け取り側の期待する定義へ変換
		if (conversionHint instanceof MethodParameter) {
			MethodParameter param = (MethodParameter) conversionHint;
			param = param.nestedIfOptional();
			if (Message.class.isAssignableFrom(param.getParameterType())) {
				param = param.nested();
			}
			Class<?> clazz = param.getNestedParameterType();
			Method parseFrom = clazz.getMethod("parseFrom", byte[].class);
			return parseFrom.invoke(null, payload);
		}
		return payload;
	}
}
```

convertFromInternalメソッドに関しては、メッセージを受け取る側の設計によるので他の方法はあると思いますが、今回は簡単な実装にしてます。

### 2. 使用するアプリケーション内でBean登録
作成したMessageConverterを使用する各アプリケーション内でBean定義します。

```Java
@Configuration
class Config {
	@Bean
	@StreamMessageConverter
	public MessageConverter messageConverter() {
		return new ProtobufMessageConverter();
	}
}
```

### 3. application.yaml(propeties)のstreamのcontent-typeにProtocol BuffersのMIMEを指定
使用するstream内のcontent-typeを認識させるために設定します。

#### 送信側

```yaml:application.yaml
spring:
  cloud:
    stream:
      bindings:
        output:
          destination: tasks
          content-type: application/x-protobuf # 作成したProtocol BuffersのMIMEを指定
          producer:
            partitionKeyExpression: "1"

```

#### 受信側

```yaml:application.yaml
spring:
  cloud:
    stream:
      bindings:
        input:
          destination: tasks
          content-type: application/x-protobuf # 作成したProtocol BuffersのMIMEを指定
```

以上でProtocol Buffers用の対応は完了です。


## 使い方
後は通常のSpring Cloud StreamとProtocol Buffersの使い方と同じになります。

#### スキーマ定義

```proto:task.proto
syntax = "proto3";

option java_multiple_files = true;
option java_package = "com.example.task";

message TaskCreated {
    string id = 1;
    string name = 2;
}

message TaskStarted {
    string id = 1;
}

message TaskDone {
    string id = 1;
}
```


#### 送信処理

```Java
@EnableBinding(Source.class)
@AllArgsConstructor
class Publisher {

	Source source;

	void create() {
		TaskCreated task = // ...
		Message<TaskCreated> message = new GenericMessage<>(task);
		source.output().send(message);
	}

	void start() {
		TaskStarted task = // ...
		Message<TaskStarted> message = new GenericMessage<>(task);
		source.output().send(message);
	}

	void done() {
		TaskDone task = // ...
		Message<TaskDone> message = new GenericMessage<>(task);
		source.output().send(message);
	}
}
```

#### 受信処理

```Java
@EnableBinding(Sink.class)
class Subscriber {

	@StreamListener(Sink.INPUT)
	void handle(TaskCreated message) {
		System.out.println(message);
	}

	@StreamListener(Sink.INPUT)
	void handle(TaskStarted message) {
		System.out.println(message);
	}

	@StreamListener(Sink.INPUT)
	void handle(TaskDone message) {
		System.out.println(message);
	}
}
```

#### 出力

```
> id: "xxx-1"
name: "Task 1"

> id: "xxx-1"

> id: "xxx-1"
```

送信側はProtocol BuffersのクラスをMessageとして送信、受信側は@StreamListenerの使用して、受け取りたいクラスを指定することで、メッセージングの実装の仕組みを意識することなく使用できます。

Spring Cloud StreamはProtocol Buffersに限らず、MessageConverterによって任意のフォーマットを使用することができるので、必要に応じアプリケーションのコードに依存しないメッセージングが可能になっていて便利ですね！


## ソースコード
今回使用したソースコードは以下にあります。
- [Github](https://github.com/disc99/spring-cloud-stream-protocol-buffers)