Spring Websocket integration with Kafka

Viewed 5651

I'm trying to send consumed Kafka data to frontend (JavaScript) via Spring-Websockets in a Spring MVC project.

To establish the communication between the server and client, I have the following.

Client (app.js)

function connect() {
    var socket = new SockJS('/kafka-data-websocket');
    stompClient = Stomp.over(socket);
    stompClient.connect({}, function (frame) {
        console.log('Connected: ' + frame);
        stompClient.send("/app/fetchData");
        stompClient.subscribe('/data/records', function (message) {
            console.log(JSON.parse(message.body).content);
        });
    });
}

Server (KafkaController.java)

@Controller
public class KafkaController {

    @MessageMapping("/fetchData")
    @SendTo("/data/records")
    public String fetchMetrics() {
        //...
    }
}

To consume data from a specific Kafka topic I'm using @KafkaListener annotation as follows:

public class KafkaReceiver {
    @KafkaListener(topics = "mytopic")
    public void receive(ConsumerRecord<?, ?> record) {
        MyRecord m = new MyRecord(new Long(record.offset()), record.key().toString(), record.value().toString());
           //...
    }
}

And I have a proper KafkaConfig class with all the necessary beans (like explained here).

How can I send data from receive method to KafkaController's fetchMetrics (and consequently to the websocket) on each incoming/consumed message?

1 Answers

You should inject SimpMessagingTemplate into the KafkaReceiver and use it from the receive() method:

 this.template.convertAndSend("/data/records", m);

See more info in the Spring Framework Reference Manual.

Related