Spring Boot RSocket: Real-time Messaging වල අනාගතය අදම දැනගමු! SC Guide

කොහොමද යාලුවනේ! Tech ලෝකේ දවසින් දවස අලුත් වෙනවා, නේද? කලින් අපි API calls කිව්වම මතක් වුණේ HTTP request/response. ඒත් දැන් ලෝකෙට ඕන වෙලා තියෙන්නේ ක්ෂණිකව, කිසිම ප්‍රමාදයකින් තොරව data එහාට මෙහාට යවන Application.

උදාහරණයක් විදියට, imagine කරන්න ඔයාගේ Chat app එකක්, Online Gaming Platform එකක්, Stock Market Dashboard එකක් නැත්නම් Real-time Analytics System එකක්. මේ වගේ තැන් වලදී HTTP වලින් විතරක් වැඩේ ගොඩදාන්න අමාරුයි, මොකද ඒක Request-Response model එකට හදපු එකක්. ඒක හරියට ඔයා කෑම ඕඩර් කරලා ඒක එනකම් බලාගෙන ඉන්නවා වගේ වැඩක්. ඒත් සමහර වෙලාවට අපිට ඕන කරන්නේ, කෑම එක හදන තැනින්ම ඉවර වුණ ගමන් අපිට notification එකක් එන එක, නැත්නම් chef ට අපේ අදහස් දෙන ගමන් කෑම එක හදන්න පුළුවන් වෙන එක. මේකට තමයි Reactive Messaging කියන concept එක ආවේ.

අද අපි කතා කරන්න යන්නේ මේ Reactive Messaging ලෝකේ ජයගන්න පුළුවන් සුපිරි technology එකක් ගැන – ඒ තමයි RSocket, විශේෂයෙන්ම Spring Boot එක්ක කොහොමද මේක වැඩ කරන්නේ කියලා. RSocket කියන්නේ සරල Request-Response වලට එහා ගිය, විවිධ ආකාරයේ communication patterns support කරන application-level protocol එකක්. මේකෙන් අපේ Application වලට අතිවිශාල Performance එකක් ලබා ගන්න පුළුවන් වගේම, Microservices අතර communication වලදී ලොකු පහසුවක් වෙනවා. ඒ කියන්නේ, මේක Spring Boot Developer කෙනෙක් විදියට ඔයාට අනිවාර්යයෙන්ම දැනගන්න ඕන දෙයක්!

මේ Guide එකෙන් අපි RSocket කියන්නේ මොකද්ද, ඒක Spring Boot එක්ක කොහොමද පාවිච්චි කරන්නේ, සහ ඒකෙන් Real-time Messaging Application එකක් හදාගන්නේ කොහොමද කියලා පියවරෙන් පියවර බලමු. ලැහැස්තිද? එහෙනම් අපි පටන් ගමු!

RSocket කියන්නේ මොකද්ද? (What is RSocket?)

සරලවම කිව්වොත්, RSocket කියන්නේ Network Communication වලට අලුත් විදියක්. මේක Message-driven protocol එකක්. HTTP වලදී වගේ Request එකක් යවලා Response එකක් එනකම් බලාගෙන ඉන්නේ නැතුව, RSocket වලින් අපිට Bidirectional, long-lived connections හදාගන්න පුළුවන්. ඒ කියන්නේ, Client ට Server ට Message යවන්නත්, Server ට Client ට Message යවන්නත් පුළුවන් එකම Connection එකක් ඇතුළේ. මේක WebSocket වලටත් වඩා advanced.

RSocket වල ප්‍රධාන විශේෂාංග:

• Reactive Streams Semantics: RSocket Reactive Streams වලට අනුව හදලා තියෙන්නේ. ඒ කියන්නේ Back-pressure support කරනවා. Server එකට handle කරන්න පුළුවන් ප්‍රමාණයට වඩා වැඩි data ප්‍රමාණයක් Client එකෙන් එවනවා නම්, Server එකට ඒ බව Client ට දැනුම් දෙන්න පුළුවන්. මේකෙන් resources waste වීම වළක්වා ගන්න පුළුවන්.
• Multiplexing: එකම TCP connection එකක් ඇතුළේ එකවර transactions ගොඩක් handle කරන්න පුළුවන්. ඒ කියන්නේ, Client එකට එකවර Server එකට Request ගොඩක් යවලා, ඒ හැම එකටම අදාළ Response වෙන වෙනම ගන්න පුළුවන්. HTTP/2 වලත් මේක තියෙනවා, ඒත් RSocket application layer එකේ මේක handle කරන නිසා වඩාත් efficient.
• Pluggable Transports: RSocket වලට TCP, WebSockets, Aeron වගේ විවිධ transport protocols පාවිච්චි කරන්න පුළුවන්. ඒකෙන් අපිට අවශ්‍ය විදියට පහසුකම් තෝරාගන්න අවස්ථාව ලැබෙනවා.
• Resume: Network connection එකක් නැති වුණොත්, Connection එක ආයෙත් ආව ගමන් කලින් තිබ්බ තැනින්ම communication එක පටන් ගන්න RSocket වලට පුළුවන්. Real-time applications වලට මේක හරිම වැදගත්.
• Load Balancing & Routing: RSocket connections අතර load balance කරන්න පුළුවන්. Microservices Architectures වලදී මේකෙන් ලොකු පහසුවක් වෙනවා.

RSocket Interaction Models:

RSocket වලදී අපිට Communication patterns 4ක් තියෙනවා. මේවා තමයි RSocket වල හරිම වැදගත් කොටස:

1. Request/Response: මේක HTTP වල වගේමයි. Client එක Request එකක් යවනවා, Server එක Response එකක් දෙනවා. (Mono<Response>)
2. Fire-and-Forget: Client එක Request එකක් යවනවා, ඒත් Server එකෙන් Response එකක් බලාපොරොත්තු වෙන්නේ නැහැ. Notifications, logging වගේ දේවල් වලට මේක සුදුසුයි. (Mono<Void>)
3. Request/Stream: Client එක Request එකක් යවනවා, Server එක ඒකට Response විදියට data Stream එකක් දෙනවා. Live data feeds, chat history වගේ දේවල් වලට මේක සුදුසුයි. (Flux<Response>)
4. Channel: මේක තමයි RSocket වල බලවත්ම pattern එක. Client එකයි Server එකයි දෙගොල්ලොම එකිනෙකාට data stream කරගන්න පුළුවන්. Real-time chat, collaborative editing වගේ දේවල් වලට මේක අත්‍යවශ්‍යයි. (Flux<Request> and Flux<Response>)

දැන් ඔයාට RSocket ගැන හොඳ අවබෝධයක් ඇති. එහෙනම් බලමු කොහොමද Spring Boot එක්ක මේක සෙට් කරගන්නේ කියලා.

Spring Boot එක්ක RSocket පටන් ගමු (Getting Started with Spring Boot and RSocket)

Spring Boot කියන්නේ Java developers ලා අතර හරිම ජනප්‍රිය Framework එකක්. RSocket integration එක Spring Boot වල හරිම පහසුවෙන් කරන්න පුළුවන්. අපි බලමු පියවරෙන් පියවර මේක කරන්නේ කොහොමද කියලා.

පියවර 1: Maven Dependency එක එකතු කරගමු

ඔයාගේ pom.xml ෆයිල් එකට මේ Dependency එක එකතු කරගන්න:

<dependency>
    <groupId>org.springframework.boot</groupId>
    <artifactId>spring-boot-starter-rsocket</artifactId>
</dependency>

පියවර 2: Server Port එක Configure කරමු

src/main/resources/application.properties ෆයිල් එකේ RSocket server එක listen කරන port එක configure කරන්න පුළුවන්. සාමාන්‍යයෙන් default port එක 7000. අපි මේක වෙනස් කරමු:

spring.rsocket.server.port=7001
spring.rsocket.server.transport=tcp

transport එක විදියට tcp හෝ websocket දාන්න පුළුවන්. Real-time chat වගේ දේකට tcp වඩාත් සුදුසුයි, මොකද ඒකෙන් direct low-level communication එකක් ලැබෙනවා.

පියවර 3: RSocket Controller එකක් හදමු

දැන් අපි සාමාන්‍ය Spring MVC වල @RestController වගේම RSocket වලට @Controller එකක් හදමු. මේකේදී අපි @MessageMapping annotation එක පාවිච්චි කරනවා. මේක හරියට REST වල @GetMapping, @PostMapping වගේමයි, හැබැයි මේක RSocket Messages වලට.

පහල තියෙන්නේ සරල Request/Response RSocket Controller එකක්. මේකෙන් Client එකෙන් 'hello' කියලා Message එකක් ආවොත්, Server එකෙන් 'Hello RSocket!' කියලා Response එකක් දෙනවා.

package com.scguide.rsocket.controller;

import org.springframework.messaging.handler.annotation.MessageMapping;
import org.springframework.stereotype.Controller;
import reactor.core.publisher.Mono;

@Controller
public class SimpleRSocketController {

    @MessageMapping("hello")
    public Mono<String> hello(String name) {
        System.out.println("Received 'hello' request with name: " + name);
        return Mono.just("Hello RSocket, " + name + "!");
    }

    @MessageMapping("fire-and-forget")
    public Mono<Void> fireAndForget(String message) {
        System.out.println("Received 'fire-and-forget' message: " + message);
        // Do something with the message, but don't send a response
        return Mono.empty();
    }
}

පියවර 4: Client එකක් හදමු (කෙටියෙන්)

RSocket Client එකක් හදන්න අපිට RSocketRequester එක පාවිච්චි කරන්න පුළුවන්. මේකෙන් Server එකට Connect වෙලා data යවන්න පුළුවන්.

package com.scguide.rsocket.client;

import org.springframework.messaging.rsocket.RSocketRequester;
import reactor.core.publisher.Mono;

public class SimpleRSocketClient {

    public static void main(String[] args) {
        RSocketRequester requester = RSocketRequester.builder()
                .tcp("localhost", 7001);

        // Request/Response example
        requester.route("hello")
                .data("SC Guide Reader")
                .retrieveMono(String.class)
                .doOnNext(response -> System.out.println("Client received: " + response))
                .block(); // block to wait for response for simple demo

        // Fire-and-Forget example
        requester.route("fire-and-forget")
                .data("This is a log message")
                .send()
                .block(); // block to ensure send completes for simple demo

        System.out.println("Fire-and-forget message sent.");
    }
}

මේක තමයි Spring Boot එක්ක RSocket පටන් ගන්න සරලම විදිය. දැන් අපි බලමු මේකෙන් කොහොමද Real-time Messaging App එකක් හදන්නේ කියලා. ඒක තමයි හරිම interesting කොටස!

Real-time Messaging App එකක් හදමු (Let's Build a Real-time Messaging App)

අපි හිතමු අපි සරල Chat Application එකක් හදනවා කියලා. මේකේදී Users ලාට එකිනෙකාට Messages යවන්න පුළුවන් වෙන්න ඕන. මේකට RSocket වල Channel model එක තමයි වඩාත් සුදුසු. ඒත් අපි අනිත් models ටිකත් කොහොමද මේකට සම්බන්ධ කරගන්නේ කියලා බලමු.

Message Data Model එක

මුලින්ම අපි Message එකක් නියෝජනය කරන්න සරල POJO (Plain Old Java Object) එකක් හදමු:

package com.scguide.rsocket.model;

import java.time.LocalDateTime;

public class Message {
    private String sender;
    private String content;
    private LocalDateTime timestamp;

    public Message() {
    }

    public Message(String sender, String content) {
        this.sender = sender;
        this.content = content;
        this.timestamp = LocalDateTime.now();
    }

    // Getters and Setters
    public String getSender() {
        return sender;
    }

    public void setSender(String sender) {
        this.sender = sender;
    }

    public String getContent() {
        return content;
    }

    public void setContent(String content) {
        this.content = content;
    }

    public LocalDateTime getTimestamp() {
        return timestamp;
    }

    public void setTimestamp(LocalDateTime timestamp) {
        this.timestamp = timestamp;
    }

    @Override
    public String toString() {
        return "[" + timestamp.toLocalTime().withNano(0) + "] " + sender + ": " + content;
    }
}

Server Side (ChatController)

අපේ Server Controller එකේදී අපි RSocket වල interaction models හතරම පාවිච්චි කරමු:

package com.scguide.rsocket.controller;

import com.scguide.rsocket.model.Message;
import org.springframework.messaging.handler.annotation.MessageMapping;
import org.springframework.stereotype.Controller;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.Sinks;

import java.time.Duration;
import java.util.ArrayList;
import java.util.List;

@Controller
public class ChatController {

    // For demonstration, simulating a message storage
    private final List<Message> messageHistory = new ArrayList<>();
    // For broadcasting new messages to all connected clients (Channel model)
    private final Sinks.Many<Message> messageSink = Sinks.many().multicast().onBackpressureBuffer();

    public ChatController() {
        // Add some initial messages for testing history
        messageHistory.add(new Message("Admin", "Welcome to SC Guide RSocket Chat!"));
        messageHistory.add(new Message("Admin", "Feel free to send messages."));
    }

    // 1. Request/Response: User login or status check
    @MessageMapping("login")
    public Mono<String> login(String username) {
        System.out.println("User " + username + " attempting to log in...");
        return Mono.just("Welcome, " + username + "! You are now connected to the chat.");
    }

    // 2. Fire-and-Forget: Sending a read receipt or a simple notification
    @MessageMapping("read.receipt")
    public Mono<Void> handleReadReceipt(String messageId) {
        System.out.println("Message with ID " + messageId + " marked as read.");
        // In a real app, you would update a database or notify sender
        return Mono.empty();
    }

    // 3. Request/Stream: Sending chat history
    @MessageMapping("chat.history")
    public Flux<Message> getChatHistory() {
        System.out.println("Serving chat history...");
        return Flux.fromIterable(messageHistory)
                .delayElements(Duration.ofMillis(100)); // Simulate delay for stream
    }

    // 4. Channel: Bi-directional chat communication
    @MessageMapping("chat.channel")
    public Flux<Message> handleChatChannel(Flux<Message> incomingMessages) {
        return incomingMessages
                .doOnNext(message -> {
                    System.out.println("Received from client: " + message);
                    messageHistory.add(message); // Store message
                    messageSink.tryEmitNext(message); // Broadcast to others
                })
                .doOnCancel(() -> System.out.println("Client disconnected from chat channel."))
                .thenMany(messageSink.asFlux()); // Continue broadcasting to the current client
    }
}

මේ Controller එකේදී අපි Sinks.Many කියන එක පාවිච්චි කරනවා, ඒක Reactive Streams වලට data publish කරන්න පාවිච්චි කරනවා. මෙතනදී අපි multicast().onBackpressureBuffer() පාවිච්චි කරලා තියෙන්නේ, සියලුම subscribers ලාට Message එක යවන්න වගේම, Client එකක back-pressure තත්වයක් ආවොත් ඒ Messages buffer කරලා තියාගන්න. chat.channel එක තමයි මේ Chat app එකේ හදවත. මේකෙන් Client එකෙන් එන Messages අරගෙන, ඒවා store කරලා, අලුත් Message එකක් ආවොත් අනිත් හැමෝටම broadcast කරනවා (messageSink.tryEmitNext(message)).

Client Side (ChatClient)

දැන් අපි Client එක හදමු. මේක Command line application එකක් විදියට හදමු.

package com.scguide.rsocket.client;

import com.scguide.rsocket.model.Message;
import org.springframework.messaging.rsocket.RSocketRequester;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.Sinks;

import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.time.Duration;
import java.util.Scanner;

public class ChatClient {

    private static String username;
    private static RSocketRequester requester;
    private static Sinks.Many<Message> messageOutboundSink;

    public static void main(String[] args) {
        Scanner scanner = new Scanner(System.in);
        System.out.print("Enter your username: ");
        username = scanner.nextLine();

        // Build RSocketRequester
        requester = RSocketRequester.builder()
                .tcp("localhost", 7001);

        // 1. Request/Response: Login
        requester.route("login")
                .data(username)
                .retrieveMono(String.class)
                .doOnNext(System.out::println)
                .block();

        // 2. Request/Stream: Get chat history
        System.out.println("\n--- Chat History ---");
        requester.route("chat.history")
                .retrieveFlux(Message.class)
                .doOnNext(System.out::println)
                .doOnComplete(() -> System.out.println("--- End of History ---\n"))
                .blockLast(); // Block until all history is received

        // 3. Channel: Start bidirectional chat
        messageOutboundSink = Sinks.many().unicast().onBackpressureBuffer();

        // Start listening for incoming messages (from server on the channel)
        requester.route("chat.channel")
                .data(messageOutboundSink.asFlux())
                .retrieveFlux(Message.class)
                .doOnNext(incomingMessage -> {
                    if (!incomingMessage.getSender().equals(username)) { // Don't print own sent messages again
                        System.out.println("" + incomingMessage);
                    }
                })
                .onErrorResume(e -> {
                    System.err.println("Chat channel error: " + e.getMessage());
                    return Mono.empty();
                })
                .subscribe(); // Subscribe to keep the connection open and receive messages

        System.out.println("Type your message and press Enter. Type 'exit' to quit.");

        // Read messages from console and send them over the channel
        BufferedReader consoleReader = new BufferedReader(new InputStreamReader(System.in));
        while (true) {
            try {
                String input = consoleReader.readLine();
                if ("exit".equalsIgnoreCase(input)) {
                    break;
                }
                Message message = new Message(username, input);
                messageOutboundSink.tryEmitNext(message);

            } catch (Exception e) {
                System.err.println("Error reading input: " + e.getMessage());
                break;
            }
        }
        scanner.close();
        System.out.println("Disconnected from chat.");
    }
}

මේ Client Code එකෙන් අපි මුලින්ම User ගේ Username එක අරගෙන Server එකට Login Request එකක් යවනවා. ඊට පස්සේ chat history එක request කරලා receive කරනවා. අන්තිමට, chat.channel එකට connect වෙලා, console එකෙන් ගන්න Messages Server එකට යවනවා වගේම, Server එකෙන් එන අලුත් Messages Console එකේ Print කරනවා. මෙතනදීත් Sinks.Many එක පාවිච්චි කරනවා Client එකෙන් Server එකට Messages යවන්න.

දැන් ඔයාට පුළුවන් මේ Project එක Build කරලා Run කරන්න. මුලින් Server එක Run කරන්න, ඊට පස්සේ Client instances කිහිපයක් Run කරලා බලන්න. ඔයාට පෙනේවි Real-time messaging කොච්චර smooth ද කියලා.

RSocket වල වාසි සහ අභියෝග (Benefits and Challenges of RSocket)

කිසිම Technology එකක් පරිපූර්ණ නැහැ. RSocket වලත් වාසි වගේම අභියෝගත් තියෙනවා. මේ දෙකම දැනගෙන සිටීම වැදගත්.

වාසි (Benefits):

• ඉහළ කාර්යසාධනය සහ කාර්යක්ෂමතාව (High Performance & Efficiency): RSocket මුලින්ම හදලම තියෙන්නේ low latency සහ high throughput වලට. ඒ කියන්නේ, මේකෙන් data හුවමාරුව හරිම වේගවත්.
• Back-pressure: මේක RSocket වල තියෙන ලොකුම වාසියක්. Server එකක් අධික data ප්‍රමාණයක් නිසා බිඳ වැටෙන එක මේකෙන් වළක්වා ගන්න පුළුවන්. ඒක හරියට පාරක traffic එක වැඩි වුණාම signal light එකෙන් vehicles flow එක regulate කරනවා වගේ වැඩක්.
• Multiplexing: එකම connection එකක් ඇතුළේ එකවර transactions ගොඩක් handle කරන්න පුළුවන් නිසා network resources save වෙනවා වගේම latency එකත් අඩු වෙනවා.
• නම්‍යශීලී Communication Models (Flexible Communication Models): Request/Response, Fire-and-Forget, Request/Stream, Channel කියන විවිධ patterns නිසා ඕනෑම Use Case එකකට සුදුසුම communication pattern එක තෝරගන්න පුළුවන්.
• Protocol Agnostic: TCP, WebSockets, Aeron වගේ ඕනෑම Transport layer එකක් මත වැඩ කරන්න පුළුවන්. මේකෙන් Future-proof solution එකක් ලැබෙනවා.
• Microservices සඳහා වඩාත් සුදුසුයි (Ideal for Microservices): Microservices අතර communication වලදී RSocket ලොකු වාසියක් වෙනවා, මොකද ඒකෙන් Service-to-Service communication වල efficiency එක වැඩි කරනවා.

අභියෝග (Challenges):

• අලුත් සංකල්පය (New Paradigm): Reactive Programming සහ RSocket වල සංකල්ප අලුත් නිසා, මේවා ඉගෙන ගන්න පොඩි කාලයක් යන්න පුළුවන්, විශේෂයෙන්ම කලින් Imperative Programming වලට පුරුදු වෙච්ච අයට.
• Debugging: Reactive Streams සහ asynchronous communication නිසා debugging කිරීම සාමාන්‍ය HTTP application එකකට වඩා ටිකක් අමාරු වෙන්න පුළුවන්. Flow එක හරියට තේරුම් ගන්න ඕන.
• Tooling: RSocket අලුත් නිසා, HTTP වලට තියෙන Postman, Swagger වගේ Tools RSocket වලට තාම ඒ තරම් develop වෙලා නැහැ. ඒත් මේවා දියුණු වෙමින් පවතිනවා.
• Browser Support: Direct RSocket client support එක Browsers වලට නැහැ. ඒත් WebSockets bridge එකක් හරහා RSocket browser වල පාවිච්චි කරන්න පුළුවන්.

ඒ වුණත්, මේ අභියෝග RSocket වල වාසි එක්ක බලද්දි පොඩි ඒවා. විශේෂයෙන්ම High-performance, Low-latency Real-time systems හදනකොට RSocket කියන්නේ game-changer කෙනෙක්.

අවසන් වචන (Conclusion)

අද අපි Spring Boot එක්ක RSocket කියන්නේ මොකද්ද, ඒකෙන් Real-time Messaging App එකක් හදන්නේ කොහොමද, ඒ වගේම ඒකේ වාසි අවාසි මොනවද කියලා ගොඩක් දේවල් කතා කළා. RSocket කියන්නේ, Web development ලෝකයේ ඉදිරියට එන බලවත්ම protocol එකක්. Reactive Programming එක්ක මේකෙන් ඔයාට හදන්න පුළුවන් Applications වලට කිසිම සීමාවක් නැහැ.

ඔයා දැන් දන්නවා RSocket කියන්නේ සරල Request-Response වලට එහා ගිය, විවිධ interaction patterns තියෙන, back-pressure වගේ දේවල් support කරන සුපිරි protocol එකක් කියලා. Spring Boot එක්ක මේක හරිම පහසුවෙන් implement කරන්න පුළුවන් කියලත් ඔයා දැක්කා.

දැන් ඔයාට පුළුවන් මේක ඔයාගේ Project වලට add කරලා බලන්න. පොඩි Proof of Concept එකක් හදලා බලන්න, ඒක කොච්චර වේගවත්ද කියලා ඔයාටම තේරේවි. මුලදී ටිකක් අමාරු වුණත්, මේක ඔයාගේ Skill set එකට අනිවාර්යයෙන්ම add කරගන්න ඕන දෙයක්. මේක අනාගතය!

මේ Guide එක ඔයාට ප්‍රයෝජනවත් වුණා නම්, අනිවාර්යයෙන්ම Share කරන්න. ඒ වගේම, ඔයාලගේ අදහස්, ප්‍රශ්න, නැත්නම් මේ වගේම වෙනත් Topic එකක් ගැන දැනගන්න ඕන නම්, පහළින් Comment කරන්න. අපි ඒ ගැනත් කතා කරමු.

ආයෙත් මේ වගේ අලුත්, වටිනා දෙයක් අරන් එනකම්, ඔයාලා හැමෝටම සුභ දවසක්!