Motivation

Protobuf doesn't support zero-copy so it's not very efficient when you aim for low latency.

Desired solution

It would be nice to consider a zero-copy format as an alternative (or replacement ?) to protobuf such as flatbuffers, capn'proto or SBE.

Motivation

Recently, computers with ARM CPU such as AWS Graviton, Apple M1 and Raspberry Pi are increasing.
By releasing binaries to Maven that run on ARM CPUs, you can run them in a wide range of environments.

Desired solution

Adding Binary Support for ARM CPUs to the Maven Repository

Considered alternatives

Additional context

When I built rsocket-rpc-protobuf version 0.3.0 as a dependency on an M1 computer with ARM, I got the following error:.

Could not find rsocket-rpc-protobuf-0.3.0-osx-aarch_64.exe (io.rsocket.rpc:rsocket-rpc-protobuf:0.3.0).
     Searched in the following locations:
         https://repo.maven.apache.org/maven2/io/rsocket/rpc/rsocket-rpc-protobuf/0.3.0/rsocket-rpc-protobuf-0.3.0-osx-aarch_64.exe

The Metadata frame for rsocket-rpc-java currently copies data when encoding it It's located here:
https://github.com/rsocket/rsocket-rpc-java/blob/master/rsocket-rpc-core/src/main/java/io/rsocket/rpc/frames/Metadata.java#L25

Update the Metadata frame internally to use the TupleByteBuf found in rsocket-java 0.12

Hi,

I have a simple Rsocket client/server application and I am trying to pass traceId in between those services. I use spring-cloud-sleuth to generate the traceId and pass it to the next service.
When I debug, client side is passing the traceId to the service properly however in the server the traceId is not visible until "doSubscribe" event.
Here is the logs shows that until "doOnSubscribe" the traceId is not printed in the logs!

. ____ _ __ _ _
/\ / ' __ _ () __ __ _ \ \ \
( ( )__ | '_ | '| | ' / ` | \ \ \
\/ )| |)| | | | | || (| | ) ) ) )
' || .__|| ||| |_, | / / / /
=========||==============|/=////
:: Spring Boot :: (v2.2.4.RELEASE)

2020-03-17 14:08:49.138 INFO [,,,] 46809 --- [ main] o.ilt.example.rpc.server.SpringBootApp : No active profile set, falling back to default profiles: default
14:08:49.138 [main] INFO o.i.example.rpc.server.SpringBootApp - No active profile set, falling back to default profiles: default
2020-03-17 14:08:49.790 INFO [,,,] 46809 --- [ main] o.s.cloud.context.scope.GenericScope : BeanFactory id=03685ab6-08d3-3b2b-8db2-3a1215573bc2
14:08:49.790 [main] INFO o.s.cloud.context.scope.GenericScope - BeanFactory id=03685ab6-08d3-3b2b-8db2-3a1215573bc2
2020-03-17 14:08:50.071 INFO [,,,] 46809 --- [ main] trationDelegate$BeanPostProcessorChecker : Bean 'org.springframework.cloud.sleuth.instrument.web.client.TraceWebClientAutoConfiguration$NettyConfiguration' of type [org.springframework.cloud.sleuth.instrument.web.client.TraceWebClientAutoConfiguration$NettyConfiguration] is not eligible for getting processed by all BeanPostProcessors (for example: not eligible for auto-proxying)
14:08:50.071 [main] INFO o.s.c.s.PostProcessorRegistrationDelegate$BeanPostProcessorChecker - Bean 'org.springframework.cloud.sleuth.instrument.web.client.TraceWebClientAutoConfiguration$NettyConfiguration' of type [org.springframework.cloud.sleuth.instrument.web.client.TraceWebClientAutoConfiguration$NettyConfiguration] is not eligible for getting processed by all BeanPostProcessors (for example: not eligible for auto-proxying)
2020-03-17 14:08:50.512 INFO [,,,] 46809 --- [ main] o.ilt.example.rpc.server.SpringBootApp : Started SpringBootApp in 2.255 seconds (JVM running for 2.902)
14:08:50.512 [main] INFO o.i.example.rpc.server.SpringBootApp - Started SpringBootApp in 2.255 seconds (JVM running for 2.902)
2020-03-17 14:08:50.513 INFO [,,,] 46809 --- [ main] o.ilt.example.rpc.server.SpringBootApp : Starting the Test Service...
14:08:50.513 [main] INFO o.i.example.rpc.server.SpringBootApp - Starting the Test Service...
2020-03-17 14:09:20.379 INFO [,,,] 46809 --- [actor-tcp-nio-2] o.i.e.rpc.server.DefaultSimpleService : requestReply:span null,
14:09:20.379 [reactor-tcp-nio-2] INFO o.i.e.r.server.DefaultSimpleService - requestReply:span null,
2020-03-17 14:09:20.380 INFO [,,,] 46809 --- [actor-tcp-nio-2] o.i.e.rpc.server.DefaultSimpleService : requestReply:got message -> hello welcome example
14:09:20.380 [reactor-tcp-nio-2] INFO o.i.e.r.server.DefaultSimpleService - requestReply:got message -> hello welcome example
14:09:20.385 [reactor-tcp-nio-2] TRACE i.rsocket.rpc.tracing.SpanSubscriber - Created span [brave.opentracing.BraveSpan@4e40be5f], with name [RequestReply], child of [brave.opentracing.BraveSpanContext$Complete@6558b030]
2020-03-17 14:09:20.390 INFO [,cbd9a15dfc54197a,cbd9a15dfc54197a,false] 46809 --- [actor-tcp-nio-2] o.i.e.rpc.server.DefaultSimpleService : doOnsubscribe:org.springframework.cloud.sleuth.instrument.reactor.ScopePassingSpanSubscriber@4127652
14:09:20.390 [reactor-tcp-nio-2] INFO o.i.e.r.server.DefaultSimpleService - doOnsubscribe:org.springframework.cloud.sleuth.instrument.reactor.ScopePassingSpanSubscriber@4127652
14:09:20.391 [reactor-tcp-nio-2] TRACE i.rsocket.rpc.tracing.SpanSubscriber - On subscribe
14:09:20.391 [reactor-tcp-nio-2] TRACE i.rsocket.rpc.tracing.SpanSubscriber - On subscribe - span continued
14:09:20.391 [reactor-tcp-nio-2] TRACE i.rsocket.rpc.tracing.SpanSubscriber - Request
14:09:20.391 [reactor-tcp-nio-2] TRACE i.rsocket.rpc.tracing.SpanSubscriber - Request - continued
14:09:20.392 [reactor-tcp-nio-2] TRACE i.rsocket.rpc.tracing.SpanSubscriber - Request after cleaning. Current span [brave.opentracing.BraveSpan@4e40be5f]
14:09:20.406 [reactor-tcp-nio-2] TRACE i.rsocket.rpc.tracing.SpanSubscriber - Cleaning up
14:09:20.406 [reactor-tcp-nio-2] TRACE i.rsocket.rpc.tracing.SpanSubscriber - Span closed

When I look at the generated code, it looks like applying the Tracing span after executing the "requestreply" function:

return service.requestReply(io.netifi.rsocket.example.SimpleRequest.parseFrom(is), metadata).map(serializer).transform(requestReply).transform(*requestReplyTrace.apply(spanContext)
*);

and when I do the some change in the code as below, then trace works fine!

 return Mono.just(payload)
                .transform(requestReplyTrace.apply(spanContext))
                .flatMap(
                    p -> {
                      com.google.protobuf.CodedInputStream is =
                          com.google.protobuf.CodedInputStream.newInstance(payload.getData());
                      try {
                        return service.requestReply(SimpleRequest.parseFrom(is), metadata);
                      } catch (IOException e) {
                        return Mono.error(e);
                      }
                    })
                .map(serializer)
                .transform(requestReply);

Can you please help this one?

Thanks

Hello,

Is it possible to bring protobuf for mac and windows too?

As far as I can see, 0.2.10 it is a last version where protobuf for macOS is available:

It would be easier to develop... Thanks!

Regards,
Max

Hi,

I am benchmarking RSocket, I noticed that RSocket is very fast with out ssl. But when I used ssl, I am seeing that it slows down roughly 7x.

This is how I am configuring my ssl, is there something that I am doing wrong?

       TcpClient client = TcpClient.create()
                .host(serverHostName)
                .port(portNo);

        if(useSsl) {
            SslContext sslContext = SslContextBuilder.forClient()
                    .trustManager(classpathStream("/ssl/rootCA.pem"))
                    .build();
            client = client.secure(spec -> {
                spec.sslContext(sslContext);
            });
        }
       ClientTransport transport = TcpClientTransport.create(client);

I am using a self created CA and a self signed certificate.

Thanks for your time.

I've been trying to get the latest version working on MacOS, but It looks like the last version of rsocket-rpc-protobuf published with osx binaries was 0.2.10:

rsocket-rpc-protobuf-0.2.10-osx-x86_64.exe

Just curious if this was intentional or maybe an oversight.

Protobuf 3.12 has added support for optional fields, and with version 3.15 this is no longer an experimental feature.

Since the compiler plugin for rsocket-rpc-java depends on an older version of protoc, the plugin currently fails when confronted with a service definition that contains optional fields.

Please upgrade protoc and protobuf dependencies to a newer version so that the plugin can handle service definitions with optional fields. Since optional fields do not change the semantics of the code generated by this plugin, this should be enough in order to support optionals.

Expected Behavior

When running the generateProto task using the latest version of io.rsocket.rpc:rsocket-rpc-core/io.rsocket.rpc:rsocket-rpc-protobuf (0.2.19) and the latest version of io.rsocket:rsocket-core (1.0.1), the generated code can be compiled without any issues.

Actual Behavior

When running the generateProto task using the latest version of io.rsocket.rpc:rsocket-rpc-core/io.rsocket.rpc:rsocket-rpc-protobuf (0.2.19) and the latest version of io.rsocket:rsocket-core (1.0.1), I get an error inside generated-sources/main/rsocketRpc/com/rsocket/rpc/Blocking...Service.java:

Cannot resolve symbol 'SwitchTransformFlux'

This is because of the following snippet:

  @java.lang.Override
  public reactor.core.publisher.Flux<io.rsocket.Payload> requestChannel(org.reactivestreams.Publisher<io.rsocket.Payload> payloads) {
    return new io.rsocket.internal.SwitchTransformFlux<io.rsocket.Payload, io.rsocket.Payload>(payloads, new java.util.function.BiFunction<io.rsocket.Payload, reactor.core.publisher.Flux<io.rsocket.Payload>, org.reactivestreams.Publisher<? extends io.rsocket.Payload>>() {
      @java.lang.Override
      public org.reactivestreams.Publisher<io.rsocket.Payload> apply(io.rsocket.Payload payload, reactor.core.publisher.Flux<io.rsocket.Payload> publisher) {
        return requestChannel(payload, publisher);
      }
    });
  }

It seems to me like the SwitchTransformFlux class got removed/replaced in version 1.0.0 of rsocket-java. When I downgrade to 0.12.1, the error is gone. None the less, this is an issue as a lot of dependencies (e.g. the spring boot rsocket starter lib) already use v1.X and therefore cannot produce valid code.

Steps to Reproduce

1. Import io.rsocket:rsocket-core:1.0.1
2. Import io.rsocket.rpc:rsocket-rpc-core:0.2.19
3. Import io.rsocket.rpc:rsocket-rpc-protobuf:0.2.19
4. Create a .proto file with a service that uses a channel

service SomeService {
  rpc channel(stream SomeRequest) returns (stream SomeResponse) {}
}

5. Generate the protobuf classes (e.g. ./gradlew generateProto)
6. Try to compile the blocking server implementation (generated-sources/main/rsocketRpc/com/rsocket/rpc/BlockingSomeServiceServer.java)

Should fail with the exception mentioned above.

Possible Solution

Either reintroduce SwitchTransformFlux or change the rsocket protoc to not use it anymore.

Your Environment

• io.rsocket:rsocket-core:1.0.1
• io.rsocket.rpc:rsocket-rpc-core:0.2.19
• io.rsocket.rpc:rsocket-rpc-protobuf:0.2.19
• com.google.protobuf:protoc:3.12.1
• com.google.protobuf:protobuf-java:3.12.1

Hello!

RSocket RPC looks very promising for me but makes me surprise this project doesn't have any release since Aug 2019. Is it abandoned?

The client never gets notified when server goes away. It just waits indefinitely.
The following unit test creates a simple server that streams 200 messages to the client. Client will call Object.wait() on server socket effectively hanging the server after receiving 2 messages and then wait for 20 more messages or it will time out in 10 seconds. The client doesn't throw any exceptions nor signals onError.

package com.example.demo;

import java.util.Date;
import java.util.Optional;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

import org.junit.Test;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;

import com.poc.protobuf.UnitRequest;
import com.poc.protobuf.UnitResponse;
import com.poc.protobuf.UnitService;
import com.poc.protobuf.UnitServiceClient;
import com.poc.protobuf.UnitServiceServer;

import io.netty.buffer.ByteBuf;
import io.rsocket.RSocket;
import io.rsocket.RSocketFactory;
import io.rsocket.rpc.rsocket.RequestHandlingRSocket;
import io.rsocket.transport.netty.client.TcpClientTransport;
import io.rsocket.transport.netty.server.NettyContextCloseable;
import io.rsocket.transport.netty.server.TcpServerTransport;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

public class DefaultSimpleServiceTests {

class DefaultSimpleService implements UnitService {

@Override
public Flux<UnitResponse> requestStream(UnitRequest unitRequest, ByteBuf metadata) {
	String command = unitRequest.getRequestCommandMessage();
	
	return Flux.range(1, 200)
			.map(i -> UnitResponse.newBuilder()
					.setMessageNumber(i)
					.setResponseMessage(i + " Srever is processsing " + command + " command")
					.build());
}
}

  @Test
  public void test1() throws Exception {
	  UnitServiceServer serviceServer = new UnitServiceServer(new DefaultSimpleService(), Optional.empty(), Optional.empty());
	  
	    NettyContextCloseable serverSocket = RSocketFactory.receive()
	        .acceptor(
	            (setup, sendingSocket) ->
	                Mono.just(new RequestHandlingRSocket(serviceServer)))
	        .transport(TcpServerTransport.create(8801))
	        .start()
	        .block();
	    
	    RSocket rSocket = RSocketFactory.connect().transport(TcpClientTransport.create(8801)).start().block();

	    UnitServiceClient client = new UnitServiceClient(rSocket);
	    
	    	CountDownLatch latch = new CountDownLatch(22);
	    	
	    	client.requestStream(UnitRequest.newBuilder().setRequestCommandMessage("Give me some data!").build())
	    			.subscribe(new Subscriber<UnitResponse>() {

				private Subscription subscription;

				@Override
				public void onSubscribe(Subscription s) {
					this.subscription = s;
					s.request(1);
				}

				@Override
				public void onNext(UnitResponse t) {
					System.out.println("Received message " + t.getResponseMessage());
					latch.countDown();
					
					if (latch.getCount() < 20) {
						System.out.println("Killing server now...");
						try {
							// this will halt the thread causing server to disappear 
							serverSocket.wait();
						} catch (InterruptedException e) {
							e.printStackTrace();
						}
					}
					
					subscription.request(1);
				}

				@Override
				public void onError(Throwable t) {
					System.out.println("Error detected! " + t);
				}

				@Override
				public void onComplete() {
					System.out.println("Stream completed!");
				}
			} );
	    	
	    	latch.await(10, TimeUnit.SECONDS);
	    	System.out.println("Finished at " + new Date().toString());
  }
}

proto file

syntax = "proto3";

package com.harris.atom.poc;

import "google/protobuf/empty.proto";

option java_package = "com.poc.protobuf";
option java_outer_classname = "UnitServiceProto";
option java_multiple_files = true;

service UnitService {

// Single Request / Streaming Response
rpc RequestStream (UnitRequest) returns (stream UnitResponse) {}
}

message UnitRequest {
string requestCommandMessage = 1;
}

message UnitResponse {
string responseMessage = 1;
int32 messageNumber = 2;
}

Actual Behaviour

At the moment, all generated server RSocket are marked with javax @Inject + @Named annotations which decrease the level of flexibility regarding using the library. From my experience, most of the injections are done through the explicit declaration of the configurations (e.g., Spring Boot configurations + beans) or modules in Dagger 2. First of all, such an aggressive injection decrease the flexibility in software testing because a usual classpath scanning will eagerly include all annotation's marked classes and inject them to test, so the user would have to hack framework for injecting mocks.

Expected Behavior

As a lib user, I want to have DI standards-free library, so I will be able to choose preferable DI solution, customize dependency injection in my application and keep it highly testable.

Proposal

As a compromise, I propose to have an additional flag for a protobuf plugin similar to what mapstruct has -> http://mapstruct.org/news/2018-07-15-mapstruct-1_3_0_Beta1-is-out-with-builder-support/#constructor-injection-for-annotation-based-component-models which make it possible to enable DI annotations on the generated classes.

With that addition, each user of RSocket-RPC-Java library will be able to turn on/off explicit dependency injection on demand

final io.netty.buffer.ByteBuf metadataBuf =  
 io.rsocket.rpc.frames.Metadata.encode(io.netty.buffer.ByteBufAllocator.DEFAULT,  
 ChannelService.SERVICE, ChannelService.METHOD_CHANNEL, metadata);

Client provided metadata is not released after being copied into metadataBuf.

I created a library that allows for compositemetadata to be used:
https://github.com/regbo/lfp-rsocket-ipc-metadata

It comes with a few other features, but I think the method used could be used here to improve current support. Have a look and LMK if you ave any questions.

It would make it easier to support RSocket-RPC-Java plugin if it is written in a more flexible way using a templating engine like mustache or more convenient for codegen like JavaPoet.

This will simplify writing things such as #26 #20 #24

Currently, the generated constructor of a server will have this code:

    if (!tracer.isPresent()) {
      this.tracer = null;
      this.requestPluginTrace = io.rsocket.rpc.tracing.Tracing.traceAsChild();
    } else {
      this.tracer = tracer.get();
      this.requestPluginTrace = io.rsocket.rpc.tracing.Tracing.traceAsChild(this.tracer, PluginsService.METHOD_REQUEST_PLUGIN, io.rsocket.rpc.tracing.Tag.of("rsocket.service", PluginsService.SERVICE), io.rsocket.rpc.tracing.Tag.of("rsocket.rpc.role", "server"), io.rsocket.rpc.tracing.Tag.of("rsocket.rpc.version", ""));
    }

The problem here is that it uses requestPluginTrace = io.rsocket.rpc.tracing.Tracing.traceAsChild();, and Tracing class loads SpanContext class from OpenTracing.

It would be nice to use java.util.function.Function.identity() instead (as you did for MicroMeter: requestPlugin = java.util.function.Function.identity()). identity() is a generic function and will not reference any classes from OpenTracing

This can help tooling bypass checks on the generated code, errorprone, for example.

It would be cool if the client stub would accept a Mono<Rsocket> in the constructor so a LoadBalancedRSocketMono could be passed in. Currently my solution is to create a new client stub instance for each request invocation which could have a noticeable memory consumption for a large number of concurrent requests.

Hi,

Can I ask where I can find the rsocket rpc examples folder ?

Currently the blocking interfaces are generated when the non-blocking interfaces are generated. This is causing issues with autowiring when the blocking interfaces are not being used.

Add a parameter to the plugin to only generate plugins when they are explicitly requested.

Error when injecting SpanContext into carrier.
I have 2 services client/server Rsocket RPC, WebFlux. I add Jaeger opentracing-spring-jaeger-cloud-starter

I try to communicate by RSocket:

rSocketClient.source()
                .flatMap(rSocket -> new IdentityProviderRSocketClient(rSocket, tracer)
                        .checkClientProfile(request))

Got it with an error in to client size:

java.lang.ClassCastException: class io.opentracing.propagation.TextMapInjectAdapter cannot be cast to class io.opentracing.propagation.TextMap (io.opentracing.propagation.TextMapInjectAdapter and io.opentracing.propagation.TextMap are in unnamed module of loader 'app')
at io.jaegertracing.internal.propagation.TextMapCodec.inject(TextMapCodec.java:37) ~[jaeger-core-1.1.0.jar:1.1.0]
at io.jaegertracing.internal.PropagationRegistry$ExceptionCatchingInjectorDecorator.inject(PropagationRegistry.java:75) ~[jaeger-core-1.1.0.jar:1.1.0]
at io.jaegertracing.internal.JaegerTracer.inject(JaegerTracer.java:208) ~[jaeger-core-1.1.0.jar:1.1.0]
at io.rsocket.ipc.tracing.SpanSubscriber.(SpanSubscriber.java:122) ~[rsocket-ipc-core-0.3.0.jar:na]
at io.rsocket.ipc.tracing.Tracing.lambda$null$0(Tracing.java:137) ~[rsocket-ipc-core-0.3.0.jar:na]
at reactor.core.publisher.Operators$LiftFunction.lambda$liftScannable$1(Operators.java:2526) ~[reactor-core-3.4.11.jar:3.4.11]
at reactor.core.publisher.MonoLift.subscribeOrReturn(MonoLift.java:41) ~[reactor-core-3.4.11.jar:3.4.11]
at reactor.core.publisher.InternalMonoOperator.subscribe(InternalMonoOperator.java:57) ~[reactor-core-3.4.11.jar:3.4.11]
at reactor.core.publisher.MonoFlatMap$FlatMapMain.onNext(MonoFlatMap.java:157) ~[reactor-core-3.4.11.jar:3.4.11]
at io.opentracing.contrib.reactor.TracedSubscriber.lambda$onNext$2(TracedSubscriber.java:69) ~[opentracing-reactor-0.1.0.jar:na]
at io.opentracing.contrib.reactor.TracedSubscriber.withActiveSpan(TracedSubscriber.java:95) ~[opentracing-reactor-0.1.0.jar:na]
...

These lines use an object of the class io.opentracing.propagation.TextMapInjectAdapter - it is not inherited from TextMap:

In the method:
<C> void inject (SpanContext var1, Format <C> var2, C var3)

It is required to pass the adapter of the TextMap class:
Format <TextMap> TEXT_MAP

The io.opentracing.contrib.spring.cloud.websocket.TextMapInjectAdapter class was probably intended to be used.
I think this is the problem. Please, help.

Hi,

I am trying to build the codes but C++ compilation part is giving some errors

Here is the errors:

Task :rsocket-rpc-protobuf:compileJava_pluginExecutableJava_pluginCpp FAILED
In file included from /Users/user/Development/test-area/rsocket-rpc-java/rsocket-rpc-protobuf/src/java_plugin/cpp/rsocket/options.pb.cc:4:
/Users/user/Development/test-area/rsocket-rpc-java/rsocket-rpc-protobuf/src/java_plugin/cpp/rsocket/options.pb.h:118:56: error: only virtual member functions can be marked 'final'
::google::protobuf::io::CodedInputStream* input) final;
^~~~~
/Users/user/Development/test-area/rsocket-rpc-java/rsocket-rpc-protobuf/src/java_plugin/cpp/rsocket/options.pb.h:120:64: error: only virtual member functions can be marked 'final'
::google::protobuf::io::CodedOutputStream* output) const final;
^~~~~
/Users/user/Development/test-area/rsocket-rpc-java/rsocket-rpc-protobuf/src/java_plugin/cpp/rsocket/options.pb.h:122:68: error: only virtual member functions can be marked 'final'
bool deterministic, ::google::protobuf::uint8* target) const final;
^~~~~
/Users/user/Development/test-area/rsocket-rpc-java/rsocket-rpc-protobuf/src/java_plugin/cpp/rsocket/options.pb.cc:11:10: fatal error: 'google/protobuf/wire_format_lite_inl.h' file not found
#include <google/protobuf/wire_format_lite_inl.h>
^~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
4 errors generated.

/Users/user/Development/test-area/rsocket-rpc-java/rsocket-rpc-protobuf/src/java_plugin/cpp/rsocket/options.pb.h:118:56: error: only virtual member functions can be marked 'final'
::google::protobuf::io::CodedInputStream* input) final;
^~~~~
/Users/user/Development/test-area/rsocket-rpc-java/rsocket-rpc-protobuf/src/java_plugin/cpp/rsocket/options.pb.h:120:64: error: only virtual member functions can be marked 'final'
::google::protobuf::io::CodedOutputStream* output) const final;
^~~~~
/Users/user/Development/test-area/rsocket-rpc-java/rsocket-rpc-protobuf/src/java_plugin/cpp/rsocket/options.pb.h:122:68: error: only virtual member functions can be marked 'final'
bool deterministic, ::google::protobuf::uint8* target) const final;

Not sure it is related to my environment

OSX
C++ - 11

Thanks

Unfortunately RHEL 6 (6.10) is still mainstream at work and the protoc plugin fails with this error:

ERROR] PROTOC FAILED: xxxxxx/target/protoc-plugins/rsocket-rpc-protobuf-0.2.19-linux-x86_64.exe: /lib64/libc.so.6: version 'GLIBC_2.14' not found (required by xxxxxx/target/protoc-plugins/rsocket-rpc-protobuf-0.2.19-linux-x86_64.exe)

--rsocketRpc_out: protoc-gen-rsocketRpc: Plugin failed with status code 1.

Protoc itself works fine.

Opentracing tracer is not wired with requester stream/channel calls

There was a new release marked 0.3.0 https://github.com/rsocket/rsocket-rpc-java/releases/tag/0.3.0 and I wanted to give it a try, but it seems that it was not uploaded to mavencentral. Last version in https://mvnrepository.com/artifact/io.rsocket.rpc/rsocket-rpc-core is 0.2.19.

Are there any plans to upload it?
Thanks
Matous

Brief introduction

Hello,
I tried to upgrade dependency from 1.0.0-RC7 to 1.0.0 but when coupled with 0.2.18 version of RSocket (i cannot use 0.2.18 because of a related issue: Issue link ) it throws an error about not existing class.

Project config

group = "pl.dwojciechowski"
version = "0.7.1"
val protobufVersion = "3.12.1"
val rsocketRpcVersion = "0.2.18"
val rsocketVersion = "1.0.0"
val coroutinesVersion = "1.3.7"
val fuelVersion = "2.2.2"
val rxJavaVersion = "3.0.4"

plugins {
    id("com.github.ben-manes.versions") version "0.28.0"
    id("org.jetbrains.intellij") version "0.4.21"
    id("com.google.protobuf") version "0.8.12"
    kotlin("jvm") version "1.3.72"
    java
    idea
}

apply(plugin = "org.jetbrains.intellij")

repositories {
    mavenCentral()
    jcenter()
    maven("https://dl.bintray.com/kittinunf/maven")
}

dependencies {
    implementation(kotlin("stdlib-jdk8"))

    implementation("org.jetbrains.kotlinx:kotlinx-coroutines-core:$coroutinesVersion")
    implementation("com.github.kittinunf.fuel", "fuel", fuelVersion)
    implementation("io.reactivex.rxjava3:rxjava:$rxJavaVersion")

    implementation("io.rsocket:rsocket-core:$rsocketVersion")
    implementation("io.rsocket:rsocket-transport-local:$rsocketVersion")
    implementation("io.rsocket:rsocket-transport-netty:$rsocketVersion")
    implementation("io.rsocket.rpc:rsocket-rpc-core:$rsocketRpcVersion")
    //Do not use implementation here, compile is needed :
    // https://github.com/JetBrains/gradle-intellij-plugin/issues/239
    // https://github.com/JetBrains/gradle-intellij-plugin/issues/456
    compile("com.google.protobuf:protobuf-java:$protobufVersion")
}

Gradle error message

C:\Users\allst\IdeaProjects\wncplugin\src\generated\main\rsocketRpc\pl\dwojciechowski\proto\files\BlockingFileServiceServer.java:92: error: cannot find symbol
    return new io.rsocket.internal.SwitchTransformFlux<io.rsocket.Payload, io.rsocket.Payload>(payloads, new java.util.function.BiFunction<io.rsocket.Payload, reactor.core.publisher.Flux<io.rsocket.Payload>, org.reactivestreams.Publisher<? extends io.rsocket.Payload>>() {
                                  ^
  symbol:   class SwitchTransformFlux
  location: package io.rsocket.internal

Generated code

  @java.lang.Override
  public reactor.core.publisher.Flux<io.rsocket.Payload> requestChannel(org.reactivestreams.Publisher<io.rsocket.Payload> payloads) {
    return new io.rsocket.internal.SwitchTransformFlux<io.rsocket.Payload, io.rsocket.Payload>(payloads, new java.util.function.BiFunction<io.rsocket.Payload, reactor.core.publisher.Flux<io.rsocket.Payload>, org.reactivestreams.Publisher<? extends io.rsocket.Payload>>() {
      @java.lang.Override
      public org.reactivestreams.Publisher<io.rsocket.Payload> apply(io.rsocket.Payload payload, reactor.core.publisher.Flux<io.rsocket.Payload> publisher) {
        return requestChannel(payload, publisher);
      }
    });
  }

Hi!

If I have a service named MyService , the current implementation generates a constructor (MyServiceServer) with 3 parameters:

1. MyService
2. Optional<MeterRegistry>
3. Optional<Tracer>

The problem is that if you exclude opentracing & micrometer from the classpath, new MyServiceServer(myService, Optional.empty(), Optional.empty()) fails at runtime because it refers to the missing classes. There is a workaround for it:

new MyServiceServer(myService, (Optional) Optional.empty(), (Optional) Optional.empty());

But it would be nice if the generator will create an additional constructor, without optionals and @Inject at all:

new MyServiceServer(myService);

and delegate to this(service, Optional.empty(), Optional.empty());

Not everyone is using opentracing and/or micrometer and it will help to not have them on classpath

rsocket-rpc was started prior to composite metadata - needs to be converted to use composite metadata. Ideally this will be backward compatible with existing code

Encoding the baggage items of a span context using DefaultMetadataEncoder.encode() can lead to an error if the baggage items are too huge to fit in the default buffer capacity (which seems to be 256).
The problem is that Tracing.mapToByteBuf uses keyLength instead of valueLength for writing value to byteBuf:

Brief introduction

Hello,
I tried to upgrade dependency from 0.2.18 to 0.2.19 but I received an error that this version could not be found. On previous version it worked fine.

I know that issue is not directly connected with code, but I could not find better place to report this issue.

Project config

val protobufVersion = "3.11.4"
val rsocketRpcVersion = "0.2.19"
val rsocketVersion = "1.0.0-RC7"
val coroutinesVersion = "1.3.5"
val fuelVersion = "2.2.2"
val rxJavaVersion = "3.0.3"

plugins {
    id("com.github.ben-manes.versions") version "0.28.0"
    id("org.jetbrains.intellij") version "0.4.21"
    id("com.google.protobuf") version "0.8.12"
    kotlin("jvm") version "1.3.72"
    java
    idea
}

apply(plugin = "org.jetbrains.intellij")

repositories {
    mavenCentral()
    jcenter()
    maven("https://dl.bintray.com/kittinunf/maven")
}

dependencies {
    implementation(kotlin("stdlib-jdk8"))

    implementation("org.jetbrains.kotlinx:kotlinx-coroutines-core:$coroutinesVersion")
    implementation("com.github.kittinunf.fuel", "fuel", fuelVersion)
    implementation("io.reactivex.rxjava3:rxjava:$rxJavaVersion")

    implementation("io.rsocket:rsocket-core:$rsocketVersion")
    implementation("io.rsocket:rsocket-transport-local:$rsocketVersion")
    implementation("io.rsocket:rsocket-transport-netty:$rsocketVersion")
    implementation("io.rsocket.rpc:rsocket-rpc-core:$rsocketRpcVersion")
    //Do not use implementation here, compile is needed :
    // https://github.com/JetBrains/gradle-intellij-plugin/issues/239
    // https://github.com/JetBrains/gradle-intellij-plugin/issues/456
    compile("com.google.protobuf:protobuf-java:$protobufVersion")
}

Gradle error message

Could not determine the dependencies of task ':prepareSandbox'.
> Could not resolve all files for configuration ':runtimeClasspath'.
   > Could not resolve io.rsocket.rpc:rsocket-rpc-core:0.2.19.
     Required by:
         project :
      > Could not resolve io.rsocket.rpc:rsocket-rpc-core:0.2.19.
         > Could not parse POM https://repo.maven.apache.org/maven2/io/rsocket/rpc/rsocket-rpc-core/0.2.19/rsocket-rpc-core-0.2.19.pom
            > Could not find io.projectreactor:reactor-bom:Dysprosium-M3.
              Searched in the following locations:
                - https://repo.maven.apache.org/maven2/io/projectreactor/reactor-bom/Dysprosium-M3/reactor-bom-Dysprosium-M3.pom
                - https://jcenter.bintray.com/io/projectreactor/reactor-bom/Dysprosium-M3/reactor-bom-Dysprosium-M3.pom
                - https://dl.bintray.com/kittinunf/maven/io/projectreactor/reactor-bom/Dysprosium-M3/reactor-bom-Dysprosium-M3.pom
                - https://cache-redirector.jetbrains.com/www.jetbrains.com/intellij-repository/releases/io/projectreactor/reactor-bom/Dysprosium-M3/reactor-bom-Dysprosium-M3.pom
                - file:/C:/Users/allst/.gradle/caches/modules-2/files-2.1/com.jetbrains.intellij.idea/ideaIC/2020.1/cbeeb1f1aebd4c9ea8fb5ab990c5904a676fc41a/ideaIC-2020.1/io.projectreactor/reactor-bom/Dysprosium-M3/ivy-Dysprosium-M3.xml
                - file:/C:/Users/allst/.gradle/caches/modules-2/files-2.1/com.jetbrains.intellij.idea/ideaIC/2020.1/cbeeb1f1aebd4c9ea8fb5ab990c5904a676fc41a/ideaIC-2020.1/reactor-bom-Dysprosium-M3-withSources-withoutAnnotations.xml
      > Could not resolve io.rsocket.rpc:rsocket-rpc-core:0.2.19.
         > Could not parse POM https://jcenter.bintray.com/io/rsocket/rpc/rsocket-rpc-core/0.2.19/rsocket-rpc-core-0.2.19.pom
            > Could not find io.projectreactor:reactor-bom:Dysprosium-M3.

Possible solution:
 - Declare repository providing the artifact, see the documentation at https://docs.gradle.org/current/userguide/declaring_repositories.html

Edit 16.05.2020

After adding following repository
maven("https://repo.spring.io/libs-milestone")

The error was changed to :

Execution failed for task ':generateProto'.
> Could not resolve all files for configuration ':protobufToolsLocator_rsocketRpc'.
   > Could not find rsocket-rpc-protobuf-0.2.19-windows-x86_64.exe (io.rsocket.rpc:rsocket-rpc-protobuf:0.2.19).
     Searched in the following locations:
         https://repo.maven.apache.org/maven2/io/rsocket/rpc/rsocket-rpc-protobuf/0.2.19/rsocket-rpc-protobuf-0.2.19-windows-x86_64.exe

Possible solution:
 - Declare repository providing the artifact, see the documentation at https://docs.gradle.org/current/userguide/declaring_repositories.html

And indeed in specified location, only artifacts for Linux are available. Windows is missing.

User Story

As an RSocket-RPC-Java user, I want to have metrics pluggable

Actual Behavior

In the current implementation, metrics are built-in, so it is hardly manageable to eliminate it from the generated code.

Expected Behavior

The current implementation provides proper abstraction and code generation that does not require a user to live with opentracing and micrometer dependencies as an built-in code. In turn, it would be much easier to have an addon to current plugin that can generate adapters for metrics and tracing that could be plugged in as in the following sample:

new TracingAwareMetricsSnapshotHandlerServerAdapter(
	new MeterAwareMetricsSnapshotHandlerServerAdapter(
    	new MetricsSnapshotHandlerServer(...),
    	meterRegistry
	)
)

in that case server/client, impl will be metrics-free code which does the same and adapters code will look like the following (example of generate server adapter):

public class MeterAwareGreeter extends AbstractRSocketService {

    private final AbstractRSocketService delegate;

    private final Function<? super Publisher<Void>, ? extends Publisher<Void>> streamGreet6MeterFunction;
    private final Function<? super Publisher<Payload>, ? extends Publisher<Payload>> streamGreet1MeterFunction;
    private final Function<? super Publisher<Payload>, ? extends Publisher<Payload>> streamGreet4MeterFunction;
    private final Function<? super Publisher<Payload>, ? extends Publisher<Payload>> streamGreet2MeterFunction;
    private final Function<? super Publisher<Payload>, ? extends Publisher<Payload>> streamGreet3MeterFunction;
    private final Function<? super Publisher<Payload>, ? extends Publisher<Payload>> streamGreet5MeterFunction;


    public MeterAwareGreeter(AbstractRSocketService delegate, MeterRegistry registry) {
        this.delegate = delegate;
        this.streamGreet6MeterFunction = Metrics.timed(registry, "rsocket.server", "service", Greeter.SERVICE, "method", Greeter.METHOD_STREAM_GREET6);
        this.streamGreet1MeterFunction = Metrics.timed(registry, "rsocket.server", "service", Greeter.SERVICE, "method", Greeter.METHOD_STREAM_GREET1);
        this.streamGreet4MeterFunction = Metrics.timed(registry, "rsocket.server", "service", Greeter.SERVICE, "method", Greeter.METHOD_STREAM_GREET4);
        this.streamGreet2MeterFunction = Metrics.timed(registry, "rsocket.server", "service", Greeter.SERVICE, "method", Greeter.METHOD_STREAM_GREET2);
        this.streamGreet3MeterFunction = Metrics.timed(registry, "rsocket.server", "service", Greeter.SERVICE, "method", Greeter.METHOD_STREAM_GREET3);
        this.streamGreet5MeterFunction = Metrics.timed(registry, "rsocket.server", "service", Greeter.SERVICE, "method", Greeter.METHOD_STREAM_GREET5);
    }

    @Override
    public Class<?> getServiceClass() {
        return delegate.getServiceClass();
    }

    @Override
    public String getService() {
        return delegate.getService();
    }


    @Override
    public Mono<Void> fireAndForget(Payload payload) {
        String methodName;

        try {
            ByteBuf metadata = payload.sliceMetadata();
            methodName = Metadata.getMethod(metadata);
        } catch (Throwable t) {
            payload.release();
            return Mono.error(t);
        }

        switch(methodName) {
            case "streamGreet6":
                return delegate.fireAndForget(payload)
                               .transform(streamGreet6MeterFunction);
            default: {
                payload.release();
                return Mono.error(new UnsupportedOperationException());
            }
        }
    }




    @Override
    public Mono<Payload> requestResponse(Payload payload) {
        String methodName;

        try {
            ByteBuf metadata = payload.sliceMetadata();
            methodName = Metadata.getMethod(metadata);
        } catch (Throwable t) {
            payload.release();
            return Mono.error(t);
        }

        switch(methodName) {
            case "streamGreet1":
                return delegate.requestResponse(payload)
                               .transform(streamGreet1MeterFunction);
            case "streamGreet4":
                return delegate.requestResponse(payload)
                               .transform(streamGreet4MeterFunction);
            default: {
                payload.release();
                return Mono.error(new UnsupportedOperationException());
            }
        }
    }
....
}

Expected Behavior

A project with both gradle-intellij-plugin and rsocket-rpc compile successfully.

Current Behavior

When both gradle-intellij-plugin and rsocket-rpc are configured in Gradle build file, build process fails on compileJava with the following message (for readability on Pastebin, and attached below)
out-error.log
out.log
Pastebin

Steps to Reproduce

1. Pull example repository github
2. install protoc on your machine (please follow this link protobuf or use a handy tool like chocolatey
3. Import project to IntelliJ IDEA
4. Execute Gradle tasks "clean" and "build"
5. Observe error message on the console 😉
6. Copy contents from "buildWithoutIntellijPlugin.gradle.kts" to "build.gradle.kts"
7. Execute Gradle tasks "clean" and "build"
8. Build should pass this time.

Context (Environment)

I am implementing opensource plugin to IntelliJ idea which will connect to another program on another machine via rsocket-rpc. I need to have generated and compiled java files o be able to use RPC goodness.

Detailed Description

I have no experience in Gradle plugin development and I am not able to find out where exactly the problem is. Maybe in the debug session, there would be a possibility to find an exact solution

The latest version of rsocket-rpc-core (0.2.19) relies on a pre-release version of reactor-bom (Dysprosium-M3 )

Motivation

• There are stable releases of reactor-bom available (e.g. Dysprosium-SR11)
• My current setup doesnt allow me to rely on pre-release libraries

Desired solution

It would be nice to have version of rsocket-rpc-corethat relies on an SR release of reactor-core 😀

Considered alternatives

Additional context