ensures InMemoryResumableFramesStore does not retain not resumable frames

Signed-off-by: Oleh Dokuka <[email protected]>

Author: OlegDokuka

rsocket-core/src/main/java/io/rsocket/resume/InMemoryResumableFramesStore.java

@@ -274,7 +274,7 @@ public void onNext(ByteBuf frame) {
 
     final CoreSubscriber<? super ByteBuf> actual = this.actual;
     if (state == 1) {
-      actual.onNext(frame.retain());
+      actual.onNext(isResumable ? frame.retainedSlice() : frame);
     } else if (!isResumable || state == 2) {
       frame.release();
     }
@@ -302,7 +302,7 @@ public void subscribe(CoreSubscriber<? super ByteBuf> actual) {
       actual.onSubscribe(this);
       synchronized (this) {
         for (final ByteBuf frame : cachedFrames) {
-          actual.onNext(frame.retain());
+          actual.onNext(frame.retainedSlice());
         }
       }