Merge pull request #9 from sbromberger/newparser

Newparser

Author: jkaye2012

REQUIRE

@@ -0,0 +1 @@
+Compat 0.4

src/Redis.jl

@@ -1,4 +1,7 @@
 module Redis
+using Compat
+
+import Base.get, Base.TcpSocket
 
 export RedisException, ConnectionException, ServerException, ProtocolException, ClientException
 export RedisConnection, SentinelConnection, TransactionConnection, SubscriptionConnection,
@@ -51,9 +54,9 @@ export sentinel_masters, sentinel_master, sentinel_slaves, sentinel_getmasteradd
        sentinel_reset, sentinel_failover, sentinel_monitor, sentinel_remove, sentinel_set
 
 include("exceptions.jl")
-include("parser.jl")
 include("connection.jl")
 include("client.jl")
+include("parser.jl")
 include("commands.jl")
 
 end

src/client.jl

@@ -46,7 +46,8 @@ function subscription_loop(conn::SubscriptionConnection, err_callback::Function)
             # socket will block until a message is received
             sleep(.1)
             nb_available(conn.socket) > 0 || continue
-            reply = parse_reply(readavailable(conn.socket))
+            l = getline(conn.socket)
+            reply = parseline(l, conn.socket)
             message = SubscriptionMessage(reply)
             if message.message_type == SubscriptionMessageType.Message
                 conn.callbacks[message.channel](message.message)
@@ -62,6 +63,8 @@ end
 macro redisfunction(command, ret_type, args...)
     func_name = esc(symbol(command))
     command = split(command, '_')
+
+
     if length(args) > 0
         return quote
             function $(func_name)(conn::RedisConnection, $(args...))

src/connection.jl

@@ -97,10 +97,3 @@ end
 function send_command(conn::RedisConnectionBase, command::String)
     write(conn.socket, command)
 end
-
-function execute_command(conn::RedisConnectionBase, command)
-    is_connected(conn) || throw(ConnectionException("Socket is disconnected"))
-    send_command(conn, pack_command(command))
-    reply = parse_reply(readavailable(conn.socket))
-    reply.response
-end

src/parser.jl

@@ -1,82 +1,69 @@
-const CRLF = "\r\n"
-
-immutable ParsedReply
-    response # The type of the response is determined by the server (Int, String, Array)
-    reply_length::Integer
+function getline(s::TcpSocket)
+    l = chomp(readline(s))
+    length(l) > 1 || throw(ProtocolException("Invalid response received: $l"))
+    return l
 end
 
-function parse_reply(reply)
-    first_crlf = search(reply, CRLF)
-    length(reply) >= 3   || throw(ProtocolException(reply))
-    first_crlf.start > 0 || throw(ProtocolException(reply))
-    reply_type = reply[1]
-    if reply_type == '+'
-        parse_simple_string_reply(reply, first_crlf)
-    elseif reply_type == '-'
-        parse_error_reply(reply, first_crlf)
-    elseif reply_type == ':'
-        parse_integer_reply(reply, first_crlf)
-    elseif reply_type == '\$'
-        parse_bulk_reply(reply, first_crlf)
-    elseif reply_type == '*'
-        parse_array_reply(reply, first_crlf)
-    else
-        throw(ProtocolException(reply))
-    end
+function parse_simple_string(l::AbstractString)
+    return l
 end
 
-# Simple strings replys extend to the first encountered CRLF
-function parse_simple_string_reply(reply, first_crlf)
-    ParsedReply(reply[2:first_crlf.start-1], first_crlf.stop)
+function parse_error(l::AbstractString)
+    throw(ServerException(l))
 end
 
-# Errors are the same as simple strings, except that their first token specifies
-# the error type
-function parse_error_reply(reply, first_crlf)
-    first_space = search(reply, ' ')
-    first_space > 0 || throw(ProtocolException(reply))
-    throw(ServerException(reply[2:first_space-1], reply[first_space+1:first_crlf.start-1]))
+function parse_integer(l)
+    return parse(Int, l)
 end
 
-# Integer replies are just ints followed by CRLF
-function parse_integer_reply(reply, first_crlf)
-    try
-        ParsedReply(parseint(reply[2:first_crlf.start-1]), first_crlf.stop)
-    catch
-        throw(ProtocolException(reply))
+function parse_bulk_string(s::TcpSocket, len::Int)
+    b = readbytes(s, len+2) # add crlf
+    if length(b) != len + 2
+        throw(ProtocolException(
+            "Bulk string read error: expected $len bytes; received $(length(b))"
+        ))
+    else
+        return join(@compat map(Char,b[1:end-2]))
     end
 end
 
-# Bulk replies specify their length and then the binary-safe string
-function parse_bulk_reply(reply, first_crlf)
-    try
-        bulk_length = parseint(reply[2:first_crlf.start-1])
-        bulk_length == -1 && return ParsedReply(nothing, first_crlf.stop)
-        reply_end = first_crlf.stop+bulk_length
-        ParsedReply(reply[first_crlf.stop+1:reply_end], reply_end+2)
-    catch
-        throw(ProtocolException(reply))
+function parse_integer(l::AbstractString)
+    return parse(Int, l)
+end
+
+function parse_array(s::TcpSocket, n::Int)
+    a = Any[]
+    for i = 1:n
+        l = getline(s)
+        r = parseline(l, s)
+        push!(a, r)
     end
+    return a
 end
 
-# Array replies specify the number of elements and then other reply types
-# for each item in length
-function parse_array_reply(reply, first_crlf)
-    try
-        array_length = parseint(reply[2:first_crlf.start-1])
-        array_length == -1 && return ParsedReply(nothing, first_crlf.stop)
-        reply = reply[first_crlf.stop+1:end]
-        reply_length = first_crlf.stop
-        response = Any[]
-        for i=1:array_length
-            parsed_element = parse_reply(reply)
-            push!(response, parsed_element.response)
-            reply_length += parsed_element.reply_length
-            reply = reply[parsed_element.reply_length+1:end]
+function parseline(l::AbstractString, s::TcpSocket)
+    reply_type = l[1]
+    reply_token = l[2:end]
+    if reply_type == '+'
+        parse_simple_string(reply_token)
+    elseif reply_type == '-'
+        parse_error(reply_token)
+    elseif reply_type == ':'
+        parse_integer(reply_token)
+    elseif reply_type == '$'
+        len = parse_integer(reply_token)
+        if len == -1
+            return nothing
+        else
+            parse_bulk_string(s, len)
+        end
+    elseif reply_type == '*'
+        len = parse_integer(reply_token)
+        if len == -1
+            return nothing
+        else
+            parse_array(s, len)
         end
-        ParsedReply(response, reply_length)
-    catch
-        throw(ProtocolException(reply))
     end
 end
 
@@ -89,6 +76,18 @@ function pack_command(command)
     packed_command
 end
 
+
+
+function execute_command(conn::RedisConnectionBase, command)
+    is_connected(conn) || throw(ConnectionException("Socket is disconnected"))
+    send_command(conn, pack_command(command))
+    l = getline(conn.socket)
+    reply = parseline(l, conn.socket)
+    return reply
+end
+
+
+
 baremodule SubscriptionMessageType
     const Message = 0
     const Pmessage = 1
@@ -100,8 +99,8 @@ immutable SubscriptionMessage
     channel::String
     message::String
 
-    function SubscriptionMessage(reply)
-        notification = reply.response
+    function SubscriptionMessage(reply::AbstractArray)
+        notification = reply
         message_type = notification[1]
         if message_type == "message"
             new(SubscriptionMessageType.Message, notification[2], notification[3])

test/redis_tests.jl

@@ -11,7 +11,7 @@ flushall(conn)
 @test set(conn, "testkey", "testvalue")
 @test get(conn, "testkey") == "testvalue"
 @test exists(conn, "testkey")
-@test keys(conn, "*") == Set({"testkey"})
+@test Redis.keys(conn, "*") == Set({"testkey"})
 @test del(conn, "testkey", "nothing", "noway") == 1
 @test get(conn, "testkey") == nothing