Hi
recently I've been thinking about what if we rewrite the whole kafka, kafkastream stack using rsocket and reactive programming? Any comments or ideas?
We already have reactive kafka.
How has been feedbacks on it?
@OlegDokuka @spencergibb when we're using backpressure and specify a max size for buffer, is there any way to signal to the producer to delay publishing new messages until the consumer is ready (instead of throwing an exception and cancelling the stream)?
routingRSocketRequester.route("responder-rr")
.data("Anders")
.retrieveFlux(String.class)
.onBackpressureBuffer(5)
.publishOn(Schedulers.boundedElastic())
.subscribe(message -> {
try {
System.out.println(message);
} catch (Exception ex) {
ex.printStackTrace();
}
});
so if you wanna get no more than 5 elements I would recomend to use the
prefetch parameter on the publishOn as in the following example
routingRSocketRequester.route("responder-rr")
.data("Anders")
.retrieveFlux(String.class)
.publishOn(Schedulers.boundedElastic(), 5)
.subscribe(message -> {
try {
System.out.println(message);
} catch (Exception ex) {
ex.printStackTrace();
}
});publishOn operator and since publishOn has to make sure the produce will not send more than the buffer size, it will request the appropriate amout when it is needed to fullfill the buffer
1 reply
Hi, I dunno if you can help me pls on rsocket net client routing to java server per https://stackoverflow.com/questions/66034186/rsocket-net-client-request-stream-routing-metadata-to-spring-boot-messagemappin and https://stackoverflow.com/questions/66157652/c-sharp-net-rsocket-client-invalid-mime-type-binary-with-java-rsocket-server
the only thing now is the other question -> the route !!
the code i am using I am getting
2021-02-12T09:51:44,503 ERROR [reactor-http-nio-3] io.rsocket.core.RSocketRequester 208 handleIncomingFrames: Unexpected error during frame handling
java.lang.IllegalStateException: Client received message for non-existent stream: 168298506, frame type: RESERVED
from the server
and
0001 Error {000}: [00000201] No handler for destination ''
at the client
so on the server we have
@MessageMapping("quotes")
Flux<Object> quotes(@Payload String message) {
return priceService.generatePrices(message);
}
on the client
sorry for the n00b route questions on net client ! ;0
await stream.ForEachAsync(quotes => Console.WriteLine($"RawDemo.OnNext===>[{quotes.Metadata}]{quotes.Data}"));
in rsocket-.net we do not have composite metadata support as of this moment
thus, you would need to do that yourselves or you would need to encode your route as a plain string
I have no idea how PdxNet works, but it is defenitelly will not be handled by spring
also, rsocket-routing is not the right place, better to ask at https://gitter.im/rsocket/rsocket-dotnet for the dotnet impl
byte[] intBytes = BitConverter.GetBytes(6);
string stringBytes = Encoding.Default.GetString(intBytes, 0, 1);
metaData = stringBytes + "quotes"; var stringclient = new RSocketClient.ForStrings(client); //A simple client that uses UTF8 strings instead of bytes.
await stringclient.RequestStream(json, metaData)
.ForEachAsync((result) =>
{
Console.WriteLine($"Result ===> {result}");
});
just want to get the binary impl going
var client = new RSocketClient(new WebSocketTransport("ws://localhost:8080/"));
await client.ConnectAsync(new RSocketOptions()
{
MetadataMimeType = "message/x.rsocket.routing.v0",
DataMimeType = "application/octet-stream"
});
Console.WriteLine("Requesting Raw Protobuf Stream...");
var route = new ReadOnlySequence<byte>(new byte[]
{
(byte) Encoding.UTF8.GetByteCount("request.stream")
}.Concat(Encoding.UTF8.GetBytes("request.stream")).ToArray());
//Make a Raw binary call just to show how it's done.
var stream = client.RequestStream(
resultmapper: result => (Data: Encoding.UTF8.GetString(result.data.ToArray()), Metadata: Encoding.UTF8.GetString(result.metadata.ToArray())),
data: new ReadOnlySequence<byte>(Encoding.UTF8.GetBytes("test")), metadata: route);
await stream.ForEachAsync(persons => Console.WriteLine($"RawDemo.OnNext===>[{persons.Metadata}]{persons.Data}"));
that what works for me
hi Oleh, I don't know if you can please have a quick look @ https://stackoverflow.com/questions/66224441/rsocket-net-for-the-3rd-message-frame-the-pipereader-readasync-is-truncated-caus
@rupweb as I mentioned before, this gitter channel is for a different project, can we move our future conversations to https://gitter.im/rsocket/rsocket-dotnet
also, what you mentioned looks like a bug
thus better to create an issue at the https://github.com/rsocket/rsocket-net/issues/new/choose
Hi Oleh, Can I use this dependency in my spring boot project?
<dependency>
<groupId>io.rsocket.routing</groupId>
<artifactId>rsocket-routing-broker-spring</artifactId>
<version>0.3.0-SNAPSHOT</version>
</dependency>I configured spring snapshot repo in my pom, but got "Not authorized" when compile.
5 replies
