Merge pull request #2 from penguineer/mvp-handler

MVP: Respond to a chat request via RabbitMQ

README.md

@@ -17,6 +17,8 @@ Configuration is done using environment variables:
 * `RMQ_PORT`: Port for RabbitMQ (default `5672`)
 * `RMQ_USER`: Username for RabbitMQ (default `guest`)
 * `RMQ_PASSWORD`: Password for RabbitMQ (default `guest`)
+* `RMQ_VHOST`: Virtual host for RabbitMQ (default `/`)
+* `RMQ_CHAT_REQUESTS`: RabbitMQ queue for chat requests (default `chat_requests`)
 * `OPENAI_API_KEY`: API key for accessing the OpenAI API
 * `OPENAI_ENDPOINT`: Base URL for the OpenAI API (default `https://api.openai.com/v1`)
 

pom.xml

@@ -77,6 +77,10 @@
 			<groupId>org.springframework.ai</groupId>
 			<artifactId>spring-ai-retry</artifactId>
 		</dependency>
+		<dependency>
+			<groupId>com.fasterxml.jackson.core</groupId>
+			<artifactId>jackson-databind</artifactId>
+		</dependency>
 	</dependencies>
 	<dependencyManagement>
 		<dependencies>

src/main/java/com/penguineering/hareairis/model/ChatRequest.java

@@ -0,0 +1,19 @@
+package com.penguineering.hareairis.model;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import java.util.UUID;
+
+@Data
+@NoArgsConstructor
+@AllArgsConstructor
+public class ChatRequest {
+    @JsonProperty("request_id")
+    private UUID requestId;
+
+    @JsonProperty("message")
+    private String message;
+}

src/main/java/com/penguineering/hareairis/model/ChatResponse.java

@@ -0,0 +1,19 @@
+package com.penguineering.hareairis.model;
+
+import com.fasterxml.jackson.annotation.JsonProperty;
+import lombok.AllArgsConstructor;
+import lombok.Data;
+import lombok.NoArgsConstructor;
+
+import java.util.UUID;
+
+@Data
+@NoArgsConstructor
+@AllArgsConstructor
+public class ChatResponse {
+    @JsonProperty("request_id")
+    private UUID requestId;
+
+    @JsonProperty("response")
+    private String response;
+}

src/main/java/com/penguineering/hareairis/rmq/ChatRequestHandler.java

@@ -0,0 +1,59 @@
+package com.penguineering.hareairis.rmq;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.penguineering.hareairis.model.ChatRequest;
+import com.penguineering.hareairis.model.ChatResponse;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.springframework.ai.chat.client.ChatClient;
+import org.springframework.amqp.core.Message;
+import org.springframework.amqp.core.MessageProperties;
+import org.springframework.amqp.rabbit.core.RabbitTemplate;
+import org.springframework.stereotype.Component;
+
+@Component
+public class ChatRequestHandler {
+
+    private static final Logger logger = LoggerFactory.getLogger(ChatRequestHandler.class);
+    private final ObjectMapper objectMapper;
+    private final ChatClient.Builder chatClientBuilder;
+    private final RabbitTemplate rabbitTemplate;
+
+    public ChatRequestHandler(ObjectMapper objectMapper,
+                              ChatClient.Builder builder,
+                              RabbitTemplate rabbitTemplate) {
+        this.objectMapper = objectMapper;
+        this.chatClientBuilder = builder;
+        this.rabbitTemplate = rabbitTemplate;
+    }
+
+    public void handleMessage(Message message) {
+        try {
+            logger.info("Received message: {}", new String(message.getBody()));
+            ChatRequest chatRequest = objectMapper.readValue(message.getBody(), ChatRequest.class);
+
+            // Extract the "reply-to" header
+            MessageProperties properties = message.getMessageProperties();
+            String replyTo = properties.getReplyTo();
+            logger.info("Reply-to header: {}", replyTo);
+
+            ChatClient chatClient = chatClientBuilder.build();
+            String response = chatClient
+                    .prompt(chatRequest.getMessage())
+                    .call()
+                    .content();
+
+            logger.info("Received response from OpenAI: {}", response);
+
+            ChatResponse chatResponse = new ChatResponse(chatRequest.getRequestId(), response);
+
+            // Convert ChatResponse to JSON
+            String jsonResponse = objectMapper.writeValueAsString(chatResponse);
+
+            // Send the response to the replyTo queue
+            rabbitTemplate.convertAndSend(replyTo, jsonResponse);
+        } catch (Exception e) {
+            logger.error("Failed to process message", e);
+        }
+    }
+}

src/main/java/com/penguineering/hareairis/rmq/RabbitMQConfig.java

@@ -0,0 +1,42 @@
+package com.penguineering.hareairis.rmq;
+
+import org.springframework.amqp.core.Queue;
+import org.springframework.amqp.rabbit.annotation.EnableRabbit;
+import org.springframework.amqp.rabbit.connection.ConnectionFactory;
+import org.springframework.amqp.rabbit.listener.SimpleMessageListenerContainer;
+import org.springframework.amqp.rabbit.listener.adapter.MessageListenerAdapter;
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.amqp.support.converter.MessageConverter;
+import org.springframework.amqp.support.converter.SimpleMessageConverter;
+
+@Configuration
+@EnableRabbit
+public class RabbitMQConfig {
+
+    @Value("${hareairis.rabbitmq.queue-chat-requests}")
+    private String queueChatRequests;
+
+    @Bean
+    public Queue chatRequestsQueue() {
+        return new Queue(queueChatRequests, true);
+    }
+
+    @Bean
+    public SimpleMessageListenerContainer chatRequestsContainer(ConnectionFactory connectionFactory,
+                                                                MessageListenerAdapter listenerAdapter) {
+        SimpleMessageListenerContainer container = new SimpleMessageListenerContainer();
+        container.setConnectionFactory(connectionFactory);
+        container.setQueueNames(queueChatRequests);
+        container.setMessageListener(listenerAdapter);
+        return container;
+    }
+
+    @Bean
+    public MessageListenerAdapter chatRequestsListenerAdapter(ChatRequestHandler handler) {
+        MessageListenerAdapter adapter = new MessageListenerAdapter(handler, "handleMessage");
+        adapter.setMessageConverter(null); // Ensure the whole message is passed
+        return adapter;
+    }
+}

src/main/resources/application.yml

@@ -20,6 +20,7 @@ spring:
     port: ${RMQ_PORT:5672}
     username: ${RMQ_USER:guest}
     password: ${RMQ_PASSWORD:guest}
+    virtual-host: ${RMQ_VHOST:/}
 
   ai:
     azure:
@@ -31,3 +32,7 @@ spring:
       backoff:
         initial-interval: 1000
         multiplier: 2
+
+hareairis:
+  rabbitmq:
+    queue-chat-requests: ${RMQ_QUEUE_CHAT_REQUESTS:chat_requests}