allow resubscriptions on client stream and channel calls (#50)

* fix channel interactions resubscriptions

Signed-off-by: Maksym Ostroverkhov <m.ostroverkhov@gmail.com>

Author: mostroverkhov

.travis.yml

@@ -22,8 +22,8 @@ jobs:
     before_install:
     - rm -r "/c/Program Files (x86)/Microsoft Visual Studio 14.0"
     - export PATH=$PATH:"/c/Program Files (x86)/Microsoft Visual Studio/2017/BuildTools/VC/Tools/MSVC/14.16.27023/bin/HostX86/x64"
-    - choco install -y adoptopenjdk8
-    - export PATH=$PATH:"/c/Program Files/AdoptOpenJDK/jdk-8.0.222.10-hotspot/bin"
+    - choco install -y adoptopenjdk8 --version 8.232.9
+    - export PATH=$PATH:"/c/Program Files/AdoptOpenJDK/jdk-8.0.232.09-hotspot/bin"
     - wget https://github.com/google/protobuf/releases/download/v3.6.1/protobuf-cpp-3.6.1.tar.gz
     - tar -xzf protobuf-cpp-3.6.1.tar.gz
     - pushd protobuf-3.6.1/cmake

rsocket-rpc-protobuf/src/java_plugin/cpp/java_generator.cpp

@@ -676,15 +676,31 @@ static void PrintClient(const ServiceDescriptor* service,
       p->Indent();
       p->Print(
           *vars,
-          "return rSocket.requestChannel($Flux$.$from$(messages).map(\n");
+          "return rSocket.requestChannel(\n");
+      p->Indent();
+      p->Print(
+          *vars,
+          "$Flux$.defer(new $Supplier$<$Publisher$<$Payload$>>() {\n\n");
+      p->Indent();
+
+      p->Print(
+          *vars,
+          "@$Override$\n"
+          "public $Publisher$<$Payload$> get() {\n");
+      p->Print(
+          *vars,
+          "final $AtomicBoolean$ once = new $AtomicBoolean$(false);\n\n");
+      p->Indent();
+            p->Print(
+                *vars,
+                "return $Flux$.$from$(messages).map(\n");
       p->Indent();
       p->Print(
           *vars,
           "new $Function$<$MessageLite$, $Payload$>() {\n");
       p->Indent();
       p->Print(
           *vars,
-          "private final $AtomicBoolean$ once = new $AtomicBoolean$(false);\n\n"
           "@$Override$\n"
           "public $Payload$ apply($MessageLite$ message) {\n");
       p->Indent();
@@ -708,6 +724,10 @@ static void PrintClient(const ServiceDescriptor* service,
       p->Outdent();
       p->Print("}\n");
       p->Outdent();
+      p->Print(" });\n");
+      p->Outdent();
+      p->Print(" }\n");
+
       if (server_streaming) {
         p->Print(
             *vars,