Okay, let's add the necessary WebSocket components to the Spring Boot backend to push notifications to clients, and then provide a basic React Native component example to receive and display them. React Native applications cannot connect directly to Kafka brokers for security and architectural reasons. A common pattern is for the backend service (our Spring Boot app) to consume from Kafka and then push relevant information to connected clients using WebSockets or native push notifications (FCM/APNS). We'll use WebSockets here. **1. Update Spring Boot Backend (Add WebSocket Support)** First, we need to add the WebSocket starter dependency and configure WebSockets. 4.0.0 org.springframework.boot spring-boot-starter-parent 3.2.5 com.example kafka-notification 0.0.1-SNAPSHOT kafka-notification Demo project for Spring Boot Kafka Notification Service 17 org.springframework.boot spring-boot-starter-web org.springframework.kafka spring-kafka com.fasterxml.jackson.core jackson-databind org.springframework.boot spring-boot-starter-websocket org.springframework.boot spring-boot-starter-test test org.springframework.kafka spring-kafka-test test org.springframework.boot spring-boot-maven-plugin # Kafka Broker Configuration spring.kafka.bootstrap-servers=localhost:9092 # Kafka Consumer Configuration spring.kafka.consumer.group-id=notification-group spring.kafka.consumer.auto-offset-reset=earliest # Configure Kafka to use JSON deserialization for values spring.kafka.consumer.value-deserializer=org.springframework.kafka.support.serializer.JsonDeserializer # Configure the JsonDeserializer to trust all packages (adjust for production) spring.kafka.consumer.properties.spring.json.trusted.packages=* # Configure the JsonDeserializer to map to our Notification class spring.kafka.consumer.properties.spring.json.value.default.type=com.example.kafkanotification.model.Notification # Kafka Producer Configuration # Configure Kafka to use JSON serialization for values spring.kafka.producer.value-serializer=org.springframework.kafka.support.serializer.JsonSerializer # Application specific properties app.kafka.notification-topic=notifications # WebSocket Configuration (Optional, defaults are often fine) # server.port=8080 # Ensure this is accessible from your RN app package com.example.kafkanotification; import org.springframework.boot.SpringApplication; import org.springframework.boot.autoconfigure.SpringBootApplication; @SpringBootApplication public class KafkaNotificationApplication { public static void main(String[] args) { SpringApplication.run(KafkaNotificationApplication.class, args); } } package com.example.kafkanotification.config; import org.apache.kafka.clients.admin.NewTopic; import org.springframework.beans.factory.annotation.Value; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.kafka.config.TopicBuilder; @Configuration public class KafkaTopicConfig { @Value("${app.kafka.notification-topic}") private String notificationTopic; @Bean public NewTopic notificationTopic() { return TopicBuilder.name(notificationTopic) .partitions(1) // Adjust partitions as needed .replicas(1) // Adjust replicas based on your Kafka setup .build(); } } package com.example.kafkanotification.config; import org.springframework.context.annotation.Configuration; import org.springframework.messaging.simp.config.MessageBrokerRegistry; import org.springframework.web.socket.config.annotation.EnableWebSocketMessageBroker; import org.springframework.web.socket.config.annotation.StompEndpointRegistry; import org.springframework.web.socket.config.annotation.WebSocketMessageBrokerConfigurer; @Configuration @EnableWebSocketMessageBroker // Enables WebSocket message handling, backed by a message broker public class WebSocketConfig implements WebSocketMessageBrokerConfigurer { @Override public void configureMessageBroker(MessageBrokerRegistry config) { // Enable a simple memory-based message broker to carry messages back to the client // on destinations prefixed with "/topic". config.enableSimpleBroker("/topic"); // Designates the "/app" prefix for messages that are bound for @MessageMapping-annotated methods. config.setApplicationDestinationPrefixes("/app"); } @Override public void registerStompEndpoints(StompEndpointRegistry registry) { // Registers the "/ws" endpoint, enabling SockJS fallback options so that alternate transports // may be used if WebSocket is not available. SockJS is used to enable fallback options // for browsers that don’t support WebSocket. // Use allowedOriginPatterns instead of allowedOrigins for more flexible CORS configuration. registry.addEndpoint("/ws").setAllowedOriginPatterns("*").withSockJS(); // You might want a non-SockJS endpoint for native clients if they support raw WebSockets better registry.addEndpoint("/ws-native").setAllowedOriginPatterns("*"); } } package com.example.kafkanotification.model; import java.time.LocalDateTime; import java.util.UUID; // Simple POJO for Notification data public class Notification { private String id; private String recipient; private String subject; private String message; private LocalDateTime timestamp; // Default constructor for Jackson Deserialization public Notification() { this.id = UUID.randomUUID().toString(); this.timestamp = LocalDateTime.now(); } public Notification(String recipient, String subject, String message) { this(); // Call default constructor to set id and timestamp this.recipient = recipient; this.subject = subject; this.message = message; } // Getters and Setters (Required for Jackson/STOMP) public String getId() { return id; } public void setId(String id) { this.id = id; } public String getRecipient() { return recipient; } public void setRecipient(String recipient) { this.recipient = recipient; } public String getSubject() { return subject; } public void setSubject(String subject) { this.subject = subject; } public String getMessage() { return message; } public void setMessage(String message) { this.message = message; } public LocalDateTime getTimestamp() { return timestamp; } public void setTimestamp(LocalDateTime timestamp) { this.timestamp = timestamp; } @Override public String toString() { return "Notification{" + "id='" + id + '\'' + ", recipient='" + recipient + '\'' + ", subject='" + subject + '\'' + ", message='" + message + '\'' + ", timestamp=" + timestamp + '}'; } } package com.example.kafkanotification.service; import com.example.kafkanotification.model.Notification; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.beans.factory.annotation.Value; import org.springframework.kafka.core.KafkaTemplate; import org.springframework.kafka.support.SendResult; import org.springframework.stereotype.Service; import java.util.concurrent.CompletableFuture; @Service public class NotificationProducerService { private static final Logger log = LoggerFactory.getLogger(NotificationProducerService.class); @Autowired private KafkaTemplate kafkaTemplate; @Value("${app.kafka.notification-topic}") private String topicName; public void sendNotification(Notification notification) { log.info("Attempting to send notification to Kafka: {}", notification); CompletableFuture> future = kafkaTemplate.send(topicName, notification.getId(), notification); future.whenComplete((result, ex) -> { if (ex == null) { log.info("Sent notification to Kafka=[{}] with offset=[{}] on partition=[{}]", notification.getId(), result.getRecordMetadata().offset(), result.getRecordMetadata().partition()); } else { log.error("Unable to send notification to Kafka=[{}] due to : {}", notification.getId(), ex.getMessage()); // Consider adding retry logic or dead-letter queue handling here } }); } } package com.example.kafkanotification.service; import com.example.kafkanotification.model.Notification; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.kafka.annotation.KafkaListener; import org.springframework.messaging.handler.annotation.Payload; import org.springframework.messaging.simp.SimpMessagingTemplate; import org.springframework.stereotype.Service; @Service public class NotificationConsumerService { private static final Logger log = LoggerFactory.getLogger(NotificationConsumerService.class); // Inject the template for sending messages over WebSocket @Autowired private SimpMessagingTemplate messagingTemplate; // Consumes messages from the topic defined in application.properties @KafkaListener(topics = "${app.kafka.notification-topic}", groupId = "${spring.kafka.consumer.group-id}", containerFactory = "kafkaListenerContainerFactory") public void listen(@Payload Notification notification) { log.info("Received notification from Kafka: {}", notification); // Process the notification (e.g., save to DB, etc.) System.out.println("Processing notification for recipient: " + notification.getRecipient()); System.out.println("Subject: " + notification.getSubject()); System.out.println("Message: " + notification.getMessage()); System.out.println("-----------------------------------------"); // Send the notification to WebSocket clients subscribed to "/topic/notifications" log.info("Sending notification to WebSocket topic /topic/notifications"); messagingTemplate.convertAndSend("/topic/notifications", notification); log.info("Notification sent to WebSocket."); } } package com.example.kafkanotification.controller; import com.example.kafkanotification.model.Notification; import com.example.kafkanotification.service.NotificationProducerService; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.http.HttpStatus; import org.springframework.http.ResponseEntity; import org.springframework.web.bind.annotation.PostMapping; import org.springframework.web.bind.annotation.RequestBody; import org.springframework.web.bind.annotation.RequestMapping; import org.springframework.web.bind.annotation.RestController; @RestController @RequestMapping("/api/notifications") public class NotificationController { @Autowired private NotificationProducerService producerService; @PostMapping("/send") public ResponseEntity sendNotification(@RequestBody Notification notificationRequest) { try { // Basic validation if (notificationRequest.getRecipient() == null || notificationRequest.getRecipient().isEmpty() || notificationRequest.getMessage() == null || notificationRequest.getMessage().isEmpty()) { return ResponseEntity.badRequest().body("Recipient and message cannot be empty."); } // Create a new Notification object to ensure ID and timestamp are set server-side Notification notificationToSend = new Notification( notificationRequest.getRecipient(), notificationRequest.getSubject() != null ? notificationRequest.getSubject() : "Default Notification Subject", notificationRequest.getMessage() ); producerService.sendNotification(notificationToSend); return ResponseEntity.ok("Notification request received and sent to Kafka."); } catch (Exception e) { // Log the exception properly in a real application System.err.println("Error sending notification: " + e.getMessage()); // Simple error logging return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR) .body("Failed to send notification: " + e.getMessage()); } } } package com.example.kafkanotification; import org.junit.jupiter.api.Test; import org.springframework.boot.test.context.SpringBootTest; @SpringBootTest class KafkaNotificationApplicationTests { @Test void contextLoads() { } } **Explanation of Backend Changes:** 1. **`pom.xml`**: Added `spring-boot-starter-websocket`. 2. **`WebSocketConfig.java`**: * `@EnableWebSocketMessageBroker`: Enables WebSocket message handling. * `configureMessageBroker`: Configures a simple message broker. Clients subscribe to destinations starting with `/topic`. Messages from the server to clients will use this prefix. `/app` is used for messages from clients to server-side `@MessageMapping` methods (not used in this example for receiving, but good practice to define). * `registerStompEndpoints`: Registers `/ws` as the WebSocket endpoint that clients will connect to. `withSockJS()` provides fallback options for environments where WebSockets might not be fully supported. `setAllowedOriginPatterns("*")` allows connections from any origin (adjust for production!). We also added `/ws-native` as an alternative without SockJS, sometimes preferred by native clients. 3. **`NotificationConsumerService.java`**: * Injected `SimpMessagingTemplate`. This is Spring's helper class for sending messages to WebSocket clients. * Inside the `listen` method (after processing the Kafka message), `messagingTemplate.convertAndSend("/topic/notifications", notification)` sends the received `Notification` object to all clients subscribed to the `/topic/notifications` destination. **2. React Native Component Example** This is a basic example. You'll need to install WebSocket/STOMP client libraries in your React Native project: bash npm install @stomp/stompjs sockjs-client text-encoding --save # or yarn add @stomp/stompjs sockjs-client text-encoding * `@stomp/stompjs`: Modern STOMP client (works over WebSockets). * `sockjs-client`: Needed if your backend uses `withSockJS()`. * `text-encoding`: Polyfill sometimes required for WebSocket libraries in React Native environments. ```javascript // src/screens/NotificationScreen.js (or wherever you place your component) import React, { useState, useEffect, useRef } from 'react'; import { View, Text, FlatList, StyleSheet, Platform, LogBox } from 'react-native'; import { Client } from '@stomp/stompjs'; import SockJS from 'sockjs-client'; import { TextEncoder, TextDecoder } from 'text-encoding'; // Polyfill // Polyfill for TextEncoder/TextDecoder if not globally available if (typeof global.TextEncoder === 'undefined') { global.TextEncoder = TextEncoder; } if (typeof global.TextDecoder === 'undefined') { global.TextDecoder = TextDecoder; } // Ignore warning related to timers in SockJS which is common in RN LogBox.ignoreLogs(['Setting a timer for a long period of time']); // --- Configuration --- // Replace with your actual backend IP/domain and port // Android Emulator: 'http://10.0.2.2:8080' // iOS Simulator or Physical Device on same Wi-Fi: 'http://:8080' const BACKEND_URL = 'http://10.0.2.2:8080'; // EXAMPLE for Android Emulator const WEBSOCKET_ENDPOINT = `${BACKEND_URL}/ws`; // Use the SockJS endpoint const NOTIFICATION_TOPIC = '/topic/notifications'; // --------------------- const NotificationScreen = () => { const [notifications, setNotifications] = useState([]); const [isConnected, setIsConnected] = useState(false); const stompClientRef = useRef(null); useEffect(() => { console.log('Setting up WebSocket connection...'); // Create STOMP client instance const client = new Client({ // Use SockJS for broader compatibility if backend uses withSockJS() webSocketFactory: () => new SockJS(WEBSOCKET_ENDPOINT), // Alternatively, for raw WebSocket (if backend has non-SockJS endpoint like /ws-native): // brokerURL: `ws://${BACKEND_URL.replace('http://', '')}/ws-native`, connectHeaders: {}, // Add any auth headers if needed debug: (str) => { console.log('STOMP DEBUG:', str); }, reconnectDelay: 5000, // Try to reconnect every 5 seconds heartbeatIncoming: 4000, heartbeatOutgoing: 4000, }); // Handle connection success client.onConnect = (frame) => { console.log('WebSocket Connected:', frame); setIsConnected(true); // Subscribe to the notification topic client.subscribe(NOTIFICATION_TOPIC, (message) => { console.log('Received message:', message.body); try { const receivedNotification = JSON.parse(message.body); // Add new notification to the beginning of the list setNotifications(prev => [receivedNotification, ...prev]); } catch (error) { console.error('Failed to parse notification:', error); } }); console.log(`Subscribed to ${NOTIFICATION_TOPIC}`); }; // Handle errors client.onStompError = (frame) => { console.error('Broker reported error:', frame.headers['message']); console.error('Additional details:', frame.body); setIsConnected(false); }; client.onWebSocketError = (event) => { console.error('WebSocket error:', event); setIsConnected(false); } client.onWebSocketClose = (event) => { console.log('WebSocket closed:', event); setIsConnected(false); } // Activate the client client.activate(); stompClientRef.current = client; // Store client in ref // Cleanup on component unmount return () => { console.log('Disconnecting WebSocket...'); if (stompClientRef.current?.connected) { stompClientRef.current.deactivate(); console.log('WebSocket deactivated.'); } else { console.log('WebSocket already inactive or not yet connected.'); } setIsConnected(false); }; }, []); // Empty dependency array ensures this runs only once on mount const renderItem = ({ item }) => ( {item.subject || 'No Subject'} To: {item.recipient} {item.message} {item.timestamp ? new Date(item.timestamp).toLocaleString() : 'No Timestamp'} ); return ( Connection Status: {isConnected ? 'Connected' : 'Disconnected'} item.id || Math.random().toString()} // Use ID or fallback ListEmptyComponent={No notifications yet.} /> ); }; const styles = StyleSheet.create({ container: { flex: 1, padding: 10, backgroundColor: '#f0f0f0', }, status: { fontSize: 16, fontWeight: 'bold', textAlign: 'center', marginBottom: 10, color: '#333', }, notificationItem: { backgroundColor: '#fff', padding: 15, marginBottom: 10, borderRadius: 5, borderWidth: 1, borderColor: '#ddd', }, subject: { fontSize: 16, fontWeight: 'bold', marginBottom: 5, }, recipient: { fontSize: 12, color: '#555', marginBottom: 5, }, message: { fontSize: 14, marginBottom: 5, }, timestamp: { fontSize: 10, color: '#888', textAlign: 'right', }, emptyText: { textAlign: 'center', marginTop: 20, fontSize: 16, color: '#666', } }); export default NotificationScreen; ``` **Explanation of React Native Code:** 1. **Imports**: Import necessary React hooks, components, and the STOMP/SockJS libraries. 2. **Polyfills**: Includes `text-encoding` polyfills which might be needed. 3. **Configuration**: Set your backend URL, WebSocket endpoint, and the STOMP topic. **Crucially, update `BACKEND_URL`** to match your Spring Boot server's IP address accessible from your simulator/device. 4. **State**: `notifications` array to hold received messages, `isConnected` boolean for connection status. 5. **`useRef`**: `stompClientRef` holds the STOMP client instance so it persists across renders and can be accessed in the cleanup function. 6. **`useEffect`**: * Runs once on component mount (`[]` dependency array). * Creates a `Client` instance from `@stomp/stompjs`. * Configures the `webSocketFactory` to use `SockJS` (matching the `/ws` endpoint in Spring Boot). * Sets up callbacks (`onConnect`, `onStompError`, `onWebSocketError`, `onWebSocketClose`). * In `onConnect`, it subscribes to `/topic/notifications`. When a message arrives, it's parsed and added to the `notifications` state. * `client.activate()` starts the connection attempt. * The `return` function is the cleanup: it deactivates the client when the component unmounts. 7. **Rendering**: * Displays the connection status. * Uses a `FlatList` to render the `notifications` array. * `renderItem` defines how each notification looks. * `ListEmptyComponent` shows a message if the list is empty. Now, when your Spring Boot application consumes a message from Kafka, it will push it via WebSocket to `/topic/notifications`, and any connected React Native clients running `NotificationScreen` will receive and display it in real-time.