I have created my first project on the github. It is Redis client protocol implementation (request-responce part only). I would like to make my sources more Kotlin style. Please, check my sources and give me some advices.
KRedis. Redis protocol implementation by Kotlin
Reply.kt
package redis.protocol.reply
import java.nio.charset.Charset
enum class ReplyType {
STATUS, ERROR, INT, BULK, MULTIBULK
}
open class Reply(val bytes: ByteArray, val type: ReplyType)
class StatusReply(bytes: ByteArray) : Reply(bytes, ReplyType.STATUS) {
companion object {
val MARKER: Char = '+'
}
constructor(data: String) : this(data.toByteArray(Charsets.UTF_8))
fun asString(): String = String(bytes, Charsets.UTF_8)
override fun toString(): String = "StatusReply(${asString()})"
}
class ErrorReply(bytes: ByteArray) : Reply(bytes, ReplyType.ERROR) {
companion object {
val MARKER: Char = '-'
}
constructor(data: String) : this(data.toByteArray(Charsets.UTF_8))
fun asString(): String = String(bytes, Charsets.UTF_8)
override fun toString(): String = "ErrorReply(${asString()})"
}
class IntegerReply(bytes: ByteArray) : Reply(bytes, ReplyType.INT) {
companion object {
val MARKER: Char = ':'
}
constructor(data: String) : this(data.toByteArray(Charsets.UTF_8))
fun asString(): String = String(bytes, Charsets.UTF_8)
@Throws(NumberFormatException::class)
fun asLong(): Long = asString().toLong()
@Throws(NumberFormatException::class)
fun asInt(): Int = asString().toInt()
override fun toString(): String = "IntegerReply(${asString()})"
}
open class BulkReply(bytes: ByteArray) : Reply(bytes, ReplyType.BULK) {
companion object {
val MARKER: Char = '$'
}
fun asByteArray(): ByteArray = bytes
fun asAsciiString(): String = String(bytes, Charsets.US_ASCII)
fun asUTF8String(): String = String(bytes, Charsets.UTF_8)
fun asString(charset: Charset = Charsets.UTF_8): String = String(bytes, charset)
override fun toString(): String = "BulkReply(${asUTF8String()})"
}
class NullBulkString() : BulkReply(byteArrayOf()) {
override fun toString(): String = "NullBulkString()"
}
class MultiBulkReply(replies: List<Reply>) : Reply(byteArrayOf(), ReplyType.MULTIBULK) {
companion object {
val MARKER: Char = '*'
}
private val _replies: List<Reply>
init {
_replies = replies
}
fun asReplyList(): List<Reply> = _replies
@Throws(IllegalArgumentException::class)
fun asStringList(charset: Charset = Charsets.UTF_8): List<String> {
if (_replies.isEmpty()) return listOf<String>()
val strings = mutableListOf<String>()
for (reply in _replies) {
when(reply) {
is StatusReply -> strings.add(reply.asString())
is IntegerReply -> strings.add(reply.asString())
is BulkReply -> strings.add(reply.asString(charset))
else -> IllegalArgumentException("Could not convert " + reply + " to a string")
}
}
return strings
}
override fun toString(): String = "MultiBulkReply(replies count = ${_replies.size})"
}
RedisProtocol.kt
package redis.protocol
import redis.protocol.reply.Reply
import redis.protocol.reply.StatusReply
import redis.protocol.reply.ErrorReply
import redis.protocol.reply.IntegerReply
import redis.protocol.reply.BulkReply
import redis.protocol.reply.NullBulkString
import redis.protocol.reply.MultiBulkReply
import java.io.BufferedInputStream
import java.io.BufferedOutputStream
import java.io.ByteArrayOutputStream
import java.io.InputStream
import java.io.OutputStream
import java.io.EOFException
import java.io.IOException
import java.net.Socket
class RedisProtocol(bis: BufferedInputStream, os: OutputStream) {
private val _is: BufferedInputStream
private val _os: OutputStream
companion object {
val CR = 13.toByte()
val LF = 10.toByte()
val NEWLINE = byteArrayOf(13, 10)
}
init {
_is = bis
_os = os
}
@Throws(IOException::class)
constructor(socket: Socket) : this(BufferedInputStream(socket.getInputStream()), BufferedOutputStream(socket.getOutputStream()))
@Throws(IOException::class, EOFException::class)
fun receive(): Reply {
val code: Int = _is.read()
if (code == -1) {
throw EOFException()
}
when (code) {
StatusReply.MARKER.toInt() -> {
return StatusReply(readSimpleElement())
}
ErrorReply.MARKER.toInt() -> {
return ErrorReply(readSimpleElement())
}
IntegerReply.MARKER.toInt() -> {
return IntegerReply(readSimpleElement())
}
BulkReply.MARKER.toInt() -> {
val (size, bytes) = readBytes()
if (size == -1)
return NullBulkString()
else
return BulkReply(bytes)
}
MultiBulkReply.MARKER.toInt() -> {
val size = String(readSimpleElement()).toInt()
val replies: List<Reply> = (1..size).map { receive() }.toList()
return MultiBulkReply(replies)
}
else -> throw IOException("Unexpected character in stream: " + code)
}
}
fun send(msg: ByteArray): Unit {
_os.write(msg)
}
@Throws(IOException::class)
private fun readSimpleElement(): ByteArray = ByteArrayOutputStream().use { boas ->
for (b: Byte in _is) {
if (b == CR) {
val lf = _is.iterator().next() // Remove byte LF from stream
if (lf == LF)
break
else
throw IOException("String that cannot contain a CR or LF character (no newlines are allowed).")
} else {
boas.write(b.toInt())
}
}
boas.toByteArray()
}
@Throws(IOException::class, NumberFormatException::class, IllegalArgumentException::class)
private fun readBytes(): Pair<Int, ByteArray> {
val size = String(readSimpleElement()).toInt()
if (size > Integer.MAX_VALUE - 8) {
throw IllegalArgumentException("Supports arrays up to ${Integer.MAX_VALUE -8 } in size")
}
if (size == -1)
return Pair(-1, byteArrayOf())
if (size < 0)
throw IllegalArgumentException("Invalid size: " + size)
var total = 0
val baos = ByteArrayOutputStream()
if (size > 0) // For correct "$0\r\n\r\n" processing
for (b: Byte in _is) {
baos.write(b.toInt())
total += 1
if (total == size) break
}
val bytes = baos.toByteArray()
baos.close()
val cr: Int = _is.read()
val lf: Int = _is.read()
if (bytes.size != size) {
throw IOException("Wrong size $size. Bytes have been read: ${bytes.size}")
}
if (cr != CR.toInt() || lf != LF.toInt()) {
throw IOException("Improper line ending: $cr, $lf")
}
return Pair(size, bytes)
}
@Throws(IOException::class, EOFException::class)
fun receiveAsync(): Reply {
synchronized (_is) {
return receive()
}
}
@Throws(IOException::class)
fun sendAsync(msg: ByteArray) {
synchronized (_os) {
send(msg)
}
_os.flush()
}
@Throws(IOException::class)
fun close() {
_is.close()
_os.close()
}
}
Command.kt
package redis.client
import redis.protocol.RedisProtocol
import java.io.ByteArrayOutputStream
enum class InsertType {
BEFORE, AFTER
}
private fun ByteArrayOutputStream.writeAsBulkString(bytes: ByteArray) {
val size: Int = bytes.size
val strSize: String = size.toString()
this.write('$'.toInt())
this.write(strSize.toByteArray(), 0, strSize.length)
this.write(RedisProtocol.NEWLINE, 0, 2)
this.write(bytes, 0, bytes.size)
this.write(RedisProtocol.NEWLINE, 0, 2)
}
private fun ByteArrayOutputStream.writeAsBulkString(value: Int) {
this.writeAsBulkString(value.toString().toByteArray(Charsets.UTF_8))
}
private fun ByteArrayOutputStream.writeAsBulkString(vararg values: String) {
for (value in values) {
this.writeAsBulkString(value.toByteArray(Charsets.UTF_8))
}
}
private fun ByteArrayOutputStream.writeAsArrayStart(arraySize: Int) {
val sArraySize = arraySize.toString()
this.write('*'.toInt())
this.write(sArraySize.toByteArray(), 0, sArraySize.length)
this.write(RedisProtocol.NEWLINE, 0, 2)
}
private fun singleCommand(cmdName: String): Command {
val baos = ByteArrayOutputStream()
val cmd = baos.use {
val size = 1 // komanda
baos.writeAsArrayStart(size)
baos.writeAsBulkString(cmdName)
baos.toByteArray()
}
return Command(cmdName, cmd)
}
private fun oneParamCommand(cmdName: String, param: ByteArray): Command {
val baos = ByteArrayOutputStream()
val cmd = baos.use {
val size = 2 // komanda + param
baos.writeAsArrayStart(size)
baos.writeAsBulkString(cmdName)
baos.writeAsBulkString(param)
baos.toByteArray()
}
return Command(cmdName, cmd)
}
private fun twoParamCommand(cmdName: String, param1: ByteArray, param2: ByteArray): Command {
val baos = ByteArrayOutputStream()
val cmd = baos.use {
val size = 3 // komanda + param
baos.writeAsArrayStart(size)
baos.writeAsBulkString(cmdName)
baos.writeAsBulkString(param1)
baos.writeAsBulkString(param2)
baos.toByteArray()
}
return Command(cmdName, cmd)
}
public fun cmdAppend(key: String, value: String): Command = cmdAppend(key.toByteArray(Charsets.UTF_8), value.toByteArray(Charsets.UTF_8))
public fun cmdAppend(key: ByteArray, value: ByteArray): Command = twoParamCommand(Command.APPEND, key, value)
public fun cmdAuth(password0: String): Command = oneParamCommand(Command.AUTH, password0.toByteArray(Charsets.UTF_8))
class Command(val name: String, val cmd: ByteArray) {
companion object Factory {
val APPEND: String = "APPEND" // Append a value to a key; Available since 2.0.0.
val AUTH: String = "AUTH" // Authenticate to the server
...
}
}
RedisClient.kt
package redis.client
import redis.protocol.RedisProtocol
import redis.protocol.reply.*
import java.net.Socket
import java.io.IOException
class RedisClient(val host: String, val port: Int, val db: Int, val passwd: String) {
lateinit var redisProtocol: RedisProtocol
constructor(host: String, port: Int, db: Int) : this(host, port, db, "")
constructor(host: String, port: Int) : this(host, port, 0, "")
@Throws(RedisException::class)
fun connect(): Boolean {
try {
redisProtocol = RedisProtocol(Socket(host, port))
if (passwd != "")
execute { cmdAuth(passwd) } // RedisException will be thrown if the ErrorReply occurs
if (db != 0)
execute { cmdSelect(db) }
return true
} catch (e: IOException) {
throw RedisException("Could not connect", e)
} finally {
}
}
@Throws(RedisException::class)
//fun execute(command: Command): Reply {
fun execute(block: () -> Command): Reply {
val command = block()
val executeReply: Reply = try {
redisProtocol.sendAsync(command.cmd)
val reply = redisProtocol.receiveAsync()
when (reply) {
is ErrorReply -> throw RedisException(reply.asString())
else -> reply
}
} catch (e: IOException) {
throw RedisException("I/O Failure: ${command.name}", e)
}
return executeReply
}
@Throws(IOException::class)
fun close() {
redisProtocol.close()
}
}