#3 Add Integration Tests

src/main/java/com/tre3p/Main.kt

@@ -16,21 +16,21 @@ import com.tre3p.storage.InMemoryKeyValueStorage
 private const val TCP_PORT = 6379
 
 fun main(args: Array<String>) {
-    prepareAndLaunchServer(args)
+    val parsedArguments = parseCliArguments(args)
+    val redisServer = prepareServer(parsedArguments)
+    redisServer.launchServer()
 }
 
-fun prepareAndLaunchServer(args: Array<String>) {
-    val cliArgs = parseCliArguments(args)
-
+fun prepareServer(args: Map<String, List<String>>): ConcurrentTcpServer {
     val mainRequestProcessor =
         MainRequestProcessor(
             RESPDecoder(),
             RESPEncoder(),
-            HandlerRouter(buildHandlerProvider(cliArgs)),
+            HandlerRouter(buildHandlerProvider(args)),
         )
 
     val redisServer = ConcurrentTcpServer(TCP_PORT, mainRequestProcessor::processRequest)
-    redisServer.launchServer()
+    return redisServer
 }
 
 private fun buildPersistenceConfig(args: Map<String, List<String>>): PersistenceConfig =

src/main/java/com/tre3p/resp/InputStreamExtensions.kt

@@ -2,7 +2,7 @@ package com.tre3p.resp
 
 import java.io.InputStream
 
-fun InputStream.readCrLfTerminatedInt(): Int {
+fun InputStream.readCrLfTerminatedElement(): String {
     val sb = StringBuilder()
 
     var byteRead: Int
@@ -22,5 +22,5 @@ fun InputStream.readCrLfTerminatedInt(): Int {
         sb.append(byteRead.toChar())
     }
 
-    return sb.toString().toInt()
+    return sb.toString()
 }

src/main/java/com/tre3p/resp/RESPDecoder.kt

@@ -7,12 +7,15 @@ open class RESPDecoder {
         when (val readByte = inputStream.read().toByte()) {
             EOF_BYTE -> null
             ASTERISK_BYTE -> parseArray(inputStream)
-            DOLLAR_BYTE -> parseString(inputStream)
+            DOLLAR_BYTE -> parseBulkString(inputStream)
+            PLUS_BYTE -> parseSimpleString(inputStream)
             else -> throw Exception("$readByte byte type isn't supported yet")
         }
 
-    private fun parseString(inputStream: InputStream): String {
-        val stringLength = inputStream.readCrLfTerminatedInt()
+    private fun parseSimpleString(inputStream: InputStream): String = inputStream.readCrLfTerminatedElement()
+
+    private fun parseBulkString(inputStream: InputStream): String {
+        val stringLength = inputStream.readCrLfTerminatedElement().toInt()
         if (stringLength < 0) return ""
 
         val readString = String(inputStream.readNBytes(stringLength))
@@ -24,7 +27,7 @@ open class RESPDecoder {
     }
 
     private fun parseArray(inputStream: InputStream): List<Any> {
-        val elementsCount = inputStream.readCrLfTerminatedInt()
+        val elementsCount = inputStream.readCrLfTerminatedElement().toInt()
         if (elementsCount < 0) return emptyList()
 
         val elements = mutableListOf<Any>()

src/main/java/com/tre3p/resp/RESPEncoder.kt

@@ -3,7 +3,6 @@ package com.tre3p.resp
 import com.tre3p.resp.types.BulkString
 import com.tre3p.resp.types.RESPArray
 import com.tre3p.resp.types.SimpleString
-import java.util.Arrays
 
 class RESPEncoder {
     fun encode(args: Any): ByteArray =
@@ -30,7 +29,6 @@ class RESPEncoder {
                     .plus(encode(it))
         }
 
-        println(Arrays.toString(initialByteArray))
         return initialByteArray
     }
 

src/main/java/com/tre3p/server/ConcurrentTcpServer.kt

@@ -20,6 +20,11 @@ class ConcurrentTcpServer(
         logger.info("TCP server started")
     }
 
+    fun stopServer() {
+        serverSocket?.close()
+        logger.info("Closing server socket")
+    }
+
     private fun launchRequestListener() {
         serverSocket ?: throw Exception("TCP server isn't launched!")
 

src/test/java/com/tre3p/BaseIntegrationTest.kt

@@ -0,0 +1,57 @@
+package com.tre3p
+
+import com.tre3p.resp.RESPDecoder
+import com.tre3p.resp.RESPEncoder
+import org.junit.jupiter.api.AfterEach
+import org.junit.jupiter.api.BeforeEach
+import java.io.ByteArrayInputStream
+import java.io.ByteArrayOutputStream
+import java.net.InetSocketAddress
+import java.net.Socket
+import java.net.SocketTimeoutException
+
+open class BaseIntegrationTest(
+    val args: Map<String, List<String>> = mapOf(),
+) {
+    private val tcpServer = prepareServer(args)
+    protected val respEncoder = RESPEncoder()
+    protected val respDecoder = RESPDecoder()
+
+    @BeforeEach
+    fun beforeEach() {
+        tcpServer.launchServer()
+    }
+
+    @AfterEach
+    fun afterEach() {
+        tcpServer.stopServer()
+    }
+
+    fun getServerConnection(): Socket {
+        val clientSocket = Socket()
+        clientSocket.connect(InetSocketAddress(6379))
+        return clientSocket
+    }
+
+    protected fun Socket.sendServerMessage(msg: ByteArray) {
+        this.getOutputStream().write(msg)
+    }
+
+    protected fun Socket.awaitServerMessage(timeoutMs: Int = 100): Any? {
+        this.soTimeout = timeoutMs
+        val buffer = ByteArray(1024)
+        val byteArrayOutputStream = ByteArrayOutputStream()
+
+        try {
+            val inputStream = this.getInputStream()
+            var bytesRead: Int
+            while (inputStream.read(buffer).also { bytesRead = it } != -1) {
+                byteArrayOutputStream.write(buffer, 0, bytesRead)
+            }
+        } catch (e: SocketTimeoutException) {
+            // Do nothing, return from method
+        }
+
+        return respDecoder.decode(ByteArrayInputStream(byteArrayOutputStream.toByteArray()))
+    }
+}

src/test/java/com/tre3p/it/EchoHandlerTest.kt

@@ -0,0 +1,20 @@
+package com.tre3p.it
+
+import com.tre3p.BaseIntegrationTest
+import com.tre3p.resp.types.RESPArray
+import com.tre3p.resp.types.SimpleString
+import org.junit.jupiter.api.Assertions
+import org.junit.jupiter.api.Test
+
+class EchoHandlerTest : BaseIntegrationTest() {
+    @Test
+    fun shouldCorrectlyHandleEchoCommand() {
+        val clientConn = getServerConnection()
+        val echoMessage = respEncoder.encode(RESPArray(listOf(SimpleString("ECHO"), SimpleString("test_message"))))
+
+        clientConn.sendServerMessage(echoMessage)
+        val serverResponse = clientConn.awaitServerMessage() as String
+
+        Assertions.assertEquals(serverResponse, "test_message")
+    }
+}

src/test/java/com/tre3p/it/GetHandlerTest.kt

@@ -0,0 +1,67 @@
+package com.tre3p.it
+
+import com.tre3p.BaseIntegrationTest
+import com.tre3p.resp.types.RESPArray
+import com.tre3p.resp.types.SimpleString
+import org.junit.jupiter.api.Assertions
+import org.junit.jupiter.api.Test
+
+class GetHandlerTest : BaseIntegrationTest() {
+    @Test
+    fun shouldReturnSetValue() {
+        val clientConn = getServerConnection()
+        val setMessage =
+            respEncoder.encode(
+                RESPArray(
+                    listOf(
+                        SimpleString("SET"),
+                        SimpleString("test_key"),
+                        SimpleString("test_value"),
+                    ),
+                ),
+            )
+        val getMessage = respEncoder.encode(RESPArray(listOf(SimpleString("GET"), SimpleString("test_key"))))
+
+        clientConn.sendServerMessage(setMessage)
+        val okResponse = clientConn.awaitServerMessage() as String
+
+        clientConn.sendServerMessage(getMessage)
+        val valueResponse = clientConn.awaitServerMessage() as String
+
+        Assertions.assertEquals(okResponse, "OK")
+        Assertions.assertEquals(valueResponse, "test_value")
+    }
+
+    @Test
+    fun shouldCorrectlyHandleExpiredValue() {
+        val clientConn = getServerConnection()
+        val setMessage =
+            respEncoder.encode(
+                RESPArray(
+                    listOf(
+                        SimpleString("SET"),
+                        SimpleString("test_key"),
+                        SimpleString("test_value"),
+                        SimpleString("px"),
+                        SimpleString("500"),
+                    ),
+                ),
+            )
+        val getMessage = respEncoder.encode(RESPArray(listOf(SimpleString("GET"), SimpleString("test_key"))))
+
+        clientConn.sendServerMessage(setMessage)
+        val okResponse = clientConn.awaitServerMessage() as String
+
+        clientConn.sendServerMessage(getMessage)
+        val valueResponse = clientConn.awaitServerMessage() as String
+
+        Assertions.assertEquals(okResponse, "OK")
+        Assertions.assertEquals(valueResponse, "test_value")
+
+        Thread.sleep(500)
+
+        clientConn.sendServerMessage(getMessage)
+        val expiredValueResponse = clientConn.awaitServerMessage() as String
+        Assertions.assertEquals("", expiredValueResponse)
+    }
+}

src/test/java/com/tre3p/it/PingHandlerTest.kt

@@ -0,0 +1,57 @@
+package com.tre3p.it
+
+import com.tre3p.BaseIntegrationTest
+import com.tre3p.resp.types.BulkString
+import com.tre3p.resp.types.RESPArray
+import org.junit.jupiter.api.Assertions
+import org.junit.jupiter.api.Test
+
+class PingHandlerTest : BaseIntegrationTest() {
+    @Test
+    fun shouldHandleSeveralCommandsFromSameConnection() {
+        val clientConn = getServerConnection()
+        val pingMessage = respEncoder.encode(RESPArray(listOf(BulkString("PING"))))
+
+        clientConn.sendServerMessage(pingMessage)
+        val firstServerResponse = clientConn.awaitServerMessage()
+
+        Assertions.assertNotNull(firstServerResponse)
+        Assertions.assertEquals((firstServerResponse as String), "PONG")
+
+        clientConn.sendServerMessage(pingMessage)
+        val secondServerResponse = clientConn.awaitServerMessage()
+
+        Assertions.assertNotNull(secondServerResponse)
+        Assertions.assertEquals((secondServerResponse as String), "PONG")
+    }
+
+    @Test
+    fun shouldRespondPongToPing() {
+        val clientConn = getServerConnection()
+        val pingMessage = respEncoder.encode(RESPArray(listOf(BulkString("PING"))))
+
+        clientConn.sendServerMessage(pingMessage)
+        val serverResponse = clientConn.awaitServerMessage()
+
+        Assertions.assertNotNull(serverResponse)
+        Assertions.assertEquals((serverResponse as String), "PONG")
+    }
+
+    @Test
+    fun shouldRespondToMultiplePings() {
+        val firstClientConn = getServerConnection()
+        val secondClientConn = getServerConnection()
+
+        val pingMessage = respEncoder.encode(RESPArray(listOf(BulkString("PING"))))
+
+        firstClientConn.sendServerMessage(pingMessage)
+        secondClientConn.sendServerMessage(pingMessage)
+        val firstServerResponse = firstClientConn.awaitServerMessage()
+        val secondServerResponse = secondClientConn.awaitServerMessage()
+
+        Assertions.assertNotNull(firstServerResponse)
+        Assertions.assertNotNull(secondServerResponse)
+        Assertions.assertEquals((firstServerResponse as String), "PONG")
+        Assertions.assertEquals((secondServerResponse as String), "PONG")
+    }
+}

src/test/java/com/tre3p/it/SetHandlerTest.kt

@@ -0,0 +1,62 @@
+package com.tre3p.it
+
+import com.tre3p.BaseIntegrationTest
+import com.tre3p.resp.types.RESPArray
+import com.tre3p.resp.types.SimpleString
+import org.junit.jupiter.api.Assertions
+import org.junit.jupiter.api.Test
+
+class SetHandlerTest : BaseIntegrationTest() {
+    @Test
+    fun shouldReturnOKOnValidSyntax() {
+        val clientConn = getServerConnection()
+        val setMessage =
+            respEncoder.encode(
+                RESPArray(
+                    listOf(
+                        SimpleString("SET"),
+                        SimpleString("test_key"),
+                        SimpleString("test_value"),
+                    ),
+                ),
+            )
+
+        clientConn.sendServerMessage(setMessage)
+        val serverResponse = clientConn.awaitServerMessage() as String
+
+        Assertions.assertEquals(serverResponse, "OK")
+    }
+
+    @Test
+    fun shouldReturnOKOnValidSyntaxWithPx() {
+        val clientConn = getServerConnection()
+        val setMessage =
+            respEncoder.encode(
+                RESPArray(
+                    listOf(
+                        SimpleString("SET"),
+                        SimpleString("test_key"),
+                        SimpleString("test_value"),
+                        SimpleString("px"),
+                        SimpleString("500"),
+                    ),
+                ),
+            )
+
+        clientConn.sendServerMessage(setMessage)
+        val serverResponse = clientConn.awaitServerMessage() as String
+
+        Assertions.assertEquals(serverResponse, "OK")
+    }
+
+    @Test
+    fun shouldReturnErrorOnInvalidSyntax() {
+        val clientConn = getServerConnection()
+        val setMessage = respEncoder.encode(RESPArray(listOf(SimpleString("SET"), SimpleString("test_key"))))
+
+        clientConn.sendServerMessage(setMessage)
+        val serverResponse = clientConn.awaitServerMessage() as String
+
+        Assertions.assertEquals(serverResponse, "Unexpected args count for SET command")
+    }
+}