refactor: improve SQL implementation and protocol handling

src/chrondb/api/sql/connection/handler.clj

@@ -0,0 +1,105 @@
+(ns chrondb.api.sql.connection.handler
+  "Functions for handling SQL client connections"
+  (:require [chrondb.util.logging :as log]
+            [chrondb.api.sql.protocol.core :as protocol]
+            [chrondb.api.sql.protocol.messages.reader :as reader]
+            [chrondb.api.sql.execution.query :as query])
+  (:import [java.io InputStream OutputStream]))
+
+(defn handle-query-message
+  "Handles a query message"
+  [storage index protocol-impl ^InputStream in ^OutputStream out]
+  (let [query-data (reader/read-query-message in)]
+    (if (:error query-data)
+      (do
+        (log/log-error (str "Error reading query: " (:error query-data)))
+        (-> protocol-impl
+            (.write-query-result out {:command "ERROR" :rows [] :data [] :columns []})))
+
+      (let [query-text (:query query-data)]
+        (log/log-info (str "Executing SQL query: " query-text))
+
+        (try
+          ;; Execute query and get results
+          (query/handle-query storage index out query-text)
+
+          ;; Retornamos true porque handle-query já envia as respostas diretamente para o output stream
+          true
+
+          (catch Exception e
+            (log/log-error (str "Error executing query: " (.getMessage e)))
+            (-> protocol-impl
+                (.write-query-result out {:command "ERROR"
+                                          :rows []
+                                          :error (.getMessage e)}))))))))
+
+(defn handle-terminate-message
+  "Handles a terminate message"
+  []
+  (log/log-info "Client requested termination")
+  {:terminate true})
+
+(defn handle-message
+  "Handles a client message based on its type"
+  [storage index protocol-impl ^InputStream in ^OutputStream out message]
+  (let [type (:type message)]
+    (try
+      (case type
+        ;; 'Q' = Simple Query
+        81 (handle-query-message storage index protocol-impl in out)
+        ;; 'X' = Terminate
+        88 (handle-terminate-message)
+        ;; Unsupported message type
+        (do
+          (log/log-info (str "Unsupported command: " (char type)))
+          ;; Discard message content
+          (let [length (reader/read-int in)
+                content-length (- length 4)
+                discard-buffer (byte-array content-length)]
+            (.read in discard-buffer))
+          ;; Send error response
+          (-> protocol-impl
+              (.write-query-result out
+                                   {:command "ERROR"
+                                    :rows []
+                                    :error (str "Unsupported command: " (char type))}))))
+      (catch Exception e
+        (log/log-error (str "Error processing message: " (.getMessage e)))
+        (-> protocol-impl
+            (.write-query-result out
+                                 {:command "ERROR"
+                                  :rows []
+                                  :error (str "Internal server error: " (.getMessage e))}))))))
+
+(defn handle-client-connection
+  "Handles a new client connection"
+  [storage index ^InputStream in ^OutputStream out]
+  (try
+    (let [protocol-impl (protocol/create-protocol)]
+      ;; Handle startup sequence
+      (doseq [action (-> protocol-impl (.handle-startup in out))]
+        (action))
+
+      ;; Process client messages until terminated or EOF
+      (loop []
+        (let [message (-> protocol-impl (.read-message in))]
+          (if message
+            (let [result (handle-message storage index protocol-impl in out message)]
+              (if-not (:terminate result)
+                (recur)
+                (log/log-info "Client session terminated normally")))
+            (log/log-info "Client disconnected")))))
+
+    (catch Exception e
+      (log/log-error (str "Error in client connection handler: " (.getMessage e))))))
+
+(defn create-connection-handler
+  "Creates a handler function for new client connections"
+  [storage index]
+  (fn [client-socket]
+    (try
+      (with-open [in (.getInputStream client-socket)
+                  out (.getOutputStream client-socket)]
+        (handle-client-connection storage index in out))
+      (catch Exception e
+        (log/log-error (str "Error handling client socket: " (.getMessage e)))))))

src/chrondb/api/sql/connection/server.clj

@@ -1,40 +1,82 @@
 (ns chrondb.api.sql.connection.server
-  "SQL server management"
-  (:require [clojure.core.async :as async]
-            [chrondb.api.sql.connection.client :as client]
-            [chrondb.util.logging :as log])
-  (:import [java.net ServerSocket]))
+  "Servidor de conexão SQL para o ChronDB"
+  (:require [chrondb.util.logging :as log]
+            [chrondb.api.sql.connection.handler :as handler])
+  (:import [java.net ServerSocket InetAddress]
+           [java.util.concurrent Executors]))
 
-(defn start-sql-server
-  "Starts a SQL server for ChronDB.
-   Parameters:
-   - storage: The storage implementation
-   - index: The index implementation
-   - port: The port number to listen on
-   Returns: The server socket"
-  [storage index port]
-  (let [server-socket (ServerSocket. port)]
-    (log/log-info (str "Starting SQL server on port " port))
-    (async/go
+(defn- create-server-socket
+  "Cria um socket de servidor na porta especificada"
+  [port]
+  (let [addr (InetAddress/getByName "localhost")
+        server-socket (ServerSocket. port 50 addr)]
+    server-socket))
+
+(defn- accept-clients
+  "Aceita conexões de clientes e processa usando o thread pool"
+  [^ServerSocket server-socket thread-pool client-handler]
+  (loop []
+    (when-not (.isClosed server-socket)
       (try
-        (while (not (.isClosed server-socket))
-          (try
-            (let [client-socket (.accept server-socket)]
-              (async/go
-                (client/handle-client storage index client-socket)))
-            (catch Exception e
-              (when-not (.isClosed server-socket)
-                (log/log-error (str "Error accepting SQL connection: " (.getMessage e)))))))
+        (let [socket (.accept server-socket)]
+          ;; Submit the client handling task to the thread pool
+          (.submit thread-pool ^Runnable #(client-handler socket)))
+        (catch java.net.SocketException e
+          (if (.isClosed server-socket)
+            (log/log-info "Server socket closed.")
+            (log/log-error (str "Socket error: " (.getMessage e)))))
         (catch Exception e
-          (log/log-error (str "Error in SQL server: " (.getMessage e))))))
-    server-socket))
+          (log/log-error (str "Error accepting client: " (.getMessage e)))))
+      ;; Continue accepting connections (fora do try)
+      (recur))))
+
+(defn start-sql-server
+  "Inicia um servidor SQL"
+  [storage index port]
+  (let [server-socket (create-server-socket port)
+        actual-port (.getLocalPort server-socket)
+        thread-pool (Executors/newCachedThreadPool)
+        client-handler (handler/create-connection-handler storage index)]
+
+    (log/log-info (str "SQL server started on port " actual-port))
+
+    ;; Start accepting connections in a separate thread
+    (let [accept-thread (Thread. #(accept-clients server-socket thread-pool client-handler))]
+      (.start accept-thread)
+
+      ;; Retornar o objeto ServerSocket para manter compatibilidade com testes
+      server-socket)))
 
 (defn stop-sql-server
-  "Stops the SQL server.
-   Parameters:
-   - server-socket: The server socket to close
-   Returns: nil"
-  [^ServerSocket server-socket]
-  (log/log-info "Stopping SQL server")
-  (when (and server-socket (not (.isClosed server-socket)))
-    (.close server-socket)))
+  "Para o servidor SQL"
+  [server]
+  (when server
+    (log/log-info "Stopping SQL server...")
+
+    ;; Se o objeto for um servidor-socket (compatibilidade com testes antigos)
+    (if (instance? java.net.ServerSocket server)
+      (do
+        (.close server)
+        (log/log-info "Server socket closed."))
+
+      ;; Se for um mapa com informações do servidor (nova implementação)
+      (let [{:keys [server-socket thread-pool]} server]
+        ;; Close the server socket to stop accepting new connections
+        (when server-socket
+          (.close server-socket)
+          (log/log-info "Server socket closed."))
+
+        ;; Shutdown the thread pool
+        (when thread-pool
+          (.shutdown thread-pool)
+          ;; Wait a bit for tasks to complete
+          (try
+            (.awaitTermination thread-pool 5 java.util.concurrent.TimeUnit/SECONDS)
+            (catch InterruptedException _
+              (log/log-warn "Interrupted while waiting for thread pool shutdown")))
+
+          ;; Force shutdown if tasks are still running
+          (when-not (.isTerminated thread-pool)
+            (.shutdownNow thread-pool)))))
+
+    (log/log-info "SQL server stopped")))

src/chrondb/api/sql/core.clj

@@ -1,12 +1,11 @@
 (ns chrondb.api.sql.core
-  "SQL server implementation for ChronDB"
+  "Core API for SQL protocol implementation"
   (:require [chrondb.util.logging :as log]
             [chrondb.api.sql.connection.server :as server]))
 
 (defn- try-start-server
-  "Attempts to start a server on the given port.
-   If the port is in use, returns nil.
-   Returns: The server socket or nil if port is in use"
+  "Tenta iniciar um servidor na porta especificada.
+   Retorna nil se a porta estiver em uso."
   [storage index port]
   (try
     (server/start-sql-server storage index port)
@@ -15,34 +14,34 @@
       nil)))
 
 (defn start-sql-server
-  "Starts a SQL server for ChronDB.
-   Parameters:
-   - storage: The storage implementation
-   - index: The index implementation (optional)
-   - port: The port number to listen on (optional, default: 5432)
-   Returns: The server socket"
+  "Inicia um servidor SQL para ChronDB.
+   Parâmetros:
+   - storage: A implementação de armazenamento
+   - index: A implementação de índice (opcional)
+   - port: O número da porta para escutar (opcional, padrão: 5432)
+   Retorna: O objeto do servidor"
   ([storage]
    (start-sql-server storage nil 5432))
   ([storage index]
    (start-sql-server storage index 5432))
   ([storage index port]
    (log/log-info (str "Starting ChronDB SQL server on port " port))
    (if (zero? port)
-     ;; If port is 0, let the system choose an available port
+     ;; Se a porta for 0, deixe o sistema escolher uma porta disponível
      (server/start-sql-server storage index port)
-     ;; Otherwise, try the specified port, or fall back to a random port if that fails
+     ;; Caso contrário, tente a porta especificada ou recorra a uma porta aleatória
      (if-let [server (try-start-server storage index port)]
        server
-       ;; If the specified port is in use, try with port 0 (system-assigned)
+       ;; Se a porta especificada estiver em uso, tente com a porta 0 (atribuída pelo sistema)
        (do
          (log/log-info "Falling back to system-assigned port")
          (server/start-sql-server storage index 0))))))
 
 (defn stop-sql-server
-  "Stops the SQL server.
-   Parameters:
-   - server-socket: The server socket to close
-   Returns: nil"
-  [server-socket]
+  "Para o servidor SQL.
+   Parâmetros:
+   - server: O objeto do servidor a ser fechado
+   Retorna: nil"
+  [server]
   (log/log-info "Stopping ChronDB SQL server")
-  (server/stop-sql-server server-socket))
+  (server/stop-sql-server server))

src/chrondb/api/sql/execution/functions.clj

@@ -20,9 +20,9 @@
           _ (log/log-info (str "Formatted aggregate result: " col-name " = " safe-result))]
       {(keyword col-name) safe-result})
     (catch Exception e
-      (log/log-error (str "Erro ao processar resultado de agregação: " (.getMessage e)
+      (log/log-error (str "Error processing aggregation result: " (.getMessage e)
                           "\n" (.printStackTrace e)))
-      {:error "Erro ao processar resultado de agregação"})))
+      {:error "Error processing aggregation result"})))
 
 (defn execute-aggregate-function
   "Executes an aggregate function on a collection of documents.
@@ -85,7 +85,7 @@
           (log/log-warn (str "Unsupported aggregate function: " function))
           nil)))
     (catch Exception e
-      (log/log-error (str "Erro ao executar função de agregação: " (.getMessage e)))
+      (log/log-error (str "Error executing aggregation function: " (.getMessage e)))
       (let [sw (java.io.StringWriter.)
             pw (java.io.PrintWriter. sw)]
         (.printStackTrace e pw)