Get Message From Kafka, send to Rsocket and Receive it from React client

Question

I am trying to send data from kafka using Spring cloud stream to Rsocket and then represent data on React

Here is my configuration.

@Configuration
public class RsocketConsumerConfiguration {
    
    @Bean
    public Sinks.Many<Data> sender(){
        return Sinks.many().multicast().directBestEffort();
    }
    

}

@Controller public class ServerController {

@Autowired
private Sinks.Many<Data> integer;

@MessageMapping("integer")
public Flux<Data> integer() {
    return  integer.asFlux();
}

@EnableBinding(IClientProcessor.class)
public class Listener {

    @Autowired
    private Sinks.Many<Data> integer;

    @StreamListener(IClientProcessor.INTEGER)
    public void integer(Data val) {
        System.out.println(val);
        integer.tryEmitNext(val);
    }

}

   let  client = new RSocketClient({
    transport: new RSocketWebSocketClient(
        {
            url: 'ws://localhost:7000/ws',
            wsCreator: (url) => new WebSocket(url),
            debug: true,
        },
        BufferEncoders,
    ),
    setup: {
        dataMimeType: "application/json",
        metadataMimeType: MESSAGE_RSOCKET_COMPOSITE_METADATA.string,
        keepAlive: 5000,
        lifetime: 60000,
    },
});

  client
            .then(rsocket => {
                console.log("Connected to rsocket");
                rsocket.requestStream({
                    metadata: Buffer.from(encodeCompositeMetadata([
                        [MESSAGE_RSOCKET_ROUTING, encodeRoute("integer")],
                    ])),
                 
                })
                    .subscribe({
                        onSubscribe: s => {
                            s.request(2147483647)
                        },
                        onNext: (p) => {
                            let newData = {
                                time: new Date(JSON.parse(p.data).time).getUTCSeconds(),
                                integer: JSON.parse(p.data).integer
                            }
                           newData.integer >100?setInteger(currentData => [newData, ...currentData]):setInt(currentData => [newData, ...currentData])
                           console.log(newData)
                        },
                        onError: (e) => console.error(e),
                        onComplete: () => console.log("Done")
                    });

spring.cloud.stream.bindings.integer.destination=integer Not able to see it in react app. Please advise. What I am doing wrong?

@meuhedetmeuhedet: Don't update the question text with a solution! It is better if you post the solution as an answer to your own question! You can also mark your own answer as accepted. It that way it is easy for readers to see that the problem is solved. Also, readers can vote on your answer. :) — Lii, Dec 30 '20 at 12:46
@meuhedetmeuhedet. I don't understand... Can't you put the code in code blocks in the answer text? — Lii, Dec 30 '20 at 13:02
The Q&A aspect is the critical point of stackoverflow. Without that it's just asking random strangers on the internet to debug your program for you. — Yuri Schimke, Dec 30 '20 at 13:23
I mean it probably what it is. Asking random stranger to debug your code. In a orginized way — meuhedet meuhedet, Dec 30 '20 at 13:29

score 1 · Answer 1 · answered Jan 25 '21 at 13:50

Given the data appears to be going directly from Kafka (via Spring) to the client, perhaps it would make more sense to stream Kafka messages via an internet-messaging broker to Internet-facing clients over WebSockets.

Disclosure: I am not the author of that article, but do work at that company where the author works. We see this use case frequently, so expect this approach may be useful.

Get Message From Kafka, send to Rsocket and Receive it from React client

1 Answers1

Linked