From 50d0c3dc43c60d3b3fe42e5586df49adb016a9cb Mon Sep 17 00:00:00 2001
From: Toshiaki Maki <tmaki@pivotal.io>
Date: Wed, 23 Sep 2020 18:53:52 +0900
Subject: [PATCH] Retain metadata for Flux payloads to avoid
 io.netty.util.IllegalReferenceCountException

---
 src/main/java/am/ik/rsocket/InteractionModel.java | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/src/main/java/am/ik/rsocket/InteractionModel.java b/src/main/java/am/ik/rsocket/InteractionModel.java
index 549eb3f..0fe15af 100644
--- a/src/main/java/am/ik/rsocket/InteractionModel.java
+++ b/src/main/java/am/ik/rsocket/InteractionModel.java
@@ -78,7 +78,7 @@ static Flux<Payload> payloadFlux(Args args) {
 			return scanToFlux(scanner) //
 					.transform(s -> args.log().map(__ -> s.log("input")).orElse(s)) //
 					.map(x -> DefaultPayload.create(Unpooled.wrappedBuffer(x.getBytes(UTF_8)),
-							args.composeMetadata().getT2()))
+							args.composeMetadata().getT2().retain()))
 					.doOnTerminate(scanner::close);
 		} else {
 			return Flux.just(DefaultPayload.create(args.data(), args.composeMetadata().getT2()));