h2o · kazuho · Mar 7, 2018 · Dec 27, 2016 · Mar 16, 2017 · Mar 16, 2017
diff --git a/CMakeLists.txt b/CMakeLists.txt
@@ -550,6 +550,7 @@ IF (WITH_MRUBY)
         lib/handler/mruby.c
         lib/handler/mruby/chunked.c
         lib/handler/mruby/http_request.c
+        lib/handler/mruby/redis.c
         lib/handler/mruby/sleep.c
         lib/handler/mruby/channel.c
         lib/handler/configurator/mruby.c)

diff --git a/examples/libh2o/redis-client.c b/examples/libh2o/redis-client.c
@@ -22,12 +22,13 @@
 #include <inttypes.h>
 #include <unistd.h>
 #include "h2o/redis.h"
+#include "hiredis.h"
 
 static h2o_loop_t *loop;
 static int exit_loop;
 const char *host;
 uint16_t port;
-h2o_redis_conn_t *conn;
+h2o_redis_client_t *client;
 
 static void usage(const char *cmd)
 {
@@ -43,7 +44,7 @@ static void dump_reply(redisReply *reply, unsigned indent)
     switch (reply->type) {
     case REDIS_REPLY_STRING:
 
-        fprintf(stderr, "string: %.*s\n", reply->len, reply->str);
+        fprintf(stderr, "string: %.*s\n", (int)reply->len, reply->str);
         break;
     case REDIS_REPLY_ARRAY:
         fprintf(stderr, "array: %zu\n", reply->elements);
@@ -58,29 +59,31 @@ static void dump_reply(redisReply *reply, unsigned indent)
         fprintf(stderr, "integer: %lld\n", reply->integer);
         break;
     case REDIS_REPLY_STATUS:
-        fprintf(stderr, "status: %.*s\n", reply->len, reply->str);
+        fprintf(stderr, "status: %.*s\n", (int)reply->len, reply->str);
         break;
     case REDIS_REPLY_ERROR:
-        fprintf(stderr, "error: %.*s\n", reply->len, reply->str);
+        fprintf(stderr, "error: %.*s\n", (int)reply->len, reply->str);
         break;
     default:
         fprintf(stderr, "invalid reply type: %d\n", reply->type);
     }
 }
 
-static void on_redis_command(redisReply *reply, void *cb_data)
+static void on_redis_command(redisReply *reply, void *cb_data, const char *errstr)
 {
-    if (reply == NULL) {
-        fprintf(stderr, "redis command failed due to some connection problems\n");
+    if (errstr != NULL) {
+        fprintf(stderr, "redis error: %s\n", errstr);
         return;
     }
-    dump_reply(reply, 0);
+    if (reply != NULL) {
+        dump_reply(reply, 0);
+    }
 }
 
 static void on_redis_connect(void)
 {
     fprintf(stderr, "connected to redis\n");
-    h2o_redis_command(conn, on_redis_command, "get server info", "INFO");
+    h2o_redis_command(client, on_redis_command, "get server info", "INFO");
 }
 
 static void on_redis_close(const char *errstr)
@@ -91,7 +94,7 @@ static void on_redis_close(const char *errstr)
         fprintf(stderr, "redis connection failure: %s\n", errstr);
         /* try to reconnect after 1 second */
         usleep(1000000);
-        h2o_redis_connect(conn, host, port);
+        h2o_redis_connect(client, host, port);
     }
 }
 
@@ -116,26 +119,26 @@ int main(int argc, char **argv)
     loop = h2o_evloop_create();
 #endif
 
-    conn = h2o_redis_create_connection(loop, sizeof(*conn));
-    conn->on_connect = on_redis_connect;
-    conn->on_close = on_redis_close;
+    client = h2o_redis_create_client(loop, sizeof(*client));
+    client->on_connect = on_redis_connect;
+    client->on_close = on_redis_close;
 
-    h2o_redis_connect(conn, host, port);
-    h2o_redis_command(conn, on_redis_command, "list all keys", "KEYS *");
+    h2o_redis_connect(client, host, port);
+    h2o_redis_command(client, on_redis_command, "list all keys", "KEYS *");
 
     while (!exit_loop) {
 #if H2O_USE_LIBUV
         uv_run(loop, UV_RUN_DEFAULT);
 #else
-        h2o_evloop_run(loop);
+        h2o_evloop_run(loop, INT32_MAX);
 #endif
     }
 
     ret = 0;
 
 Exit:
 
-    h2o_redis_free(conn);
+    h2o_redis_free(client);
 
     if (loop != NULL) {
 #if H2O_USE_LIBUV

diff --git a/h2o.xcodeproj/project.pbxproj b/h2o.xcodeproj/project.pbxproj
@@ -14,6 +14,7 @@
 		0812AB231D7FCFEB00004F23 /* hiredis.h in Headers */ = {isa = PBXBuildFile; fileRef = 0812AB1F1D7FCFEB00004F23 /* hiredis.h */; };
 		0812AB2B1D7FD54700004F23 /* read.c in Sources */ = {isa = PBXBuildFile; fileRef = 0812AB291D7FD54700004F23 /* read.c */; };
 		0812AB2C1D7FD54700004F23 /* read.h in Headers */ = {isa = PBXBuildFile; fileRef = 0812AB2A1D7FD54700004F23 /* read.h */; };
+		0812174E1E07B89600712F36 /* redis.c in Sources */ = {isa = PBXBuildFile; fileRef = 0812174C1E07B89600712F36 /* redis.c */; };
 		084FC7C11D54B90D00E89F66 /* http2_debug_state.c in Sources */ = {isa = PBXBuildFile; fileRef = 084FC7C01D54B90D00E89F66 /* http2_debug_state.c */; };
 		084FC7C51D54BB9200E89F66 /* http2_debug_state.c in Sources */ = {isa = PBXBuildFile; fileRef = 084FC7C31D54BB9200E89F66 /* http2_debug_state.c */; };
 		08790DDA1D80153600A04BC1 /* net.c in Sources */ = {isa = PBXBuildFile; fileRef = 08790DD91D80153600A04BC1 /* net.c */; };
@@ -498,6 +499,7 @@
 		0812AB1F1D7FCFEB00004F23 /* hiredis.h */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.c.h; path = hiredis.h; sourceTree = "<group>"; };
 		0812AB291D7FD54700004F23 /* read.c */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.c.c; path = read.c; sourceTree = "<group>"; };
 		0812AB2A1D7FD54700004F23 /* read.h */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.c.h; path = read.h; sourceTree = "<group>"; };
+		0812174C1E07B89600712F36 /* redis.c */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.c.c; path = redis.c; sourceTree = "<group>"; };
 		084FC7C01D54B90D00E89F66 /* http2_debug_state.c */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.c.c; path = http2_debug_state.c; sourceTree = "<group>"; };
 		084FC7C31D54BB9200E89F66 /* http2_debug_state.c */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.c.c; path = http2_debug_state.c; sourceTree = "<group>"; };
 		08790DD91D80153600A04BC1 /* net.c */ = {isa = PBXFileReference; fileEncoding = 4; lastKnownFileType = sourcecode.c.c; path = net.c; sourceTree = "<group>"; };
@@ -1057,6 +1059,7 @@
 				08E9CC4D1E41F6660049DD26 /* embedded.c.h */,
 				100A55141C30C5BC00C4E3E0 /* chunked.c */,
 				100A550E1C2BB15100C4E3E0 /* http_request.c */,
+				0812174C1E07B89600712F36 /* redis.c */,
 				7D0285C81EF422D40094292B /* sleep.c */,
 				7D9FA5381FC323AC00189F88 /* channel.c */,
 			);
@@ -2630,6 +2633,7 @@
 				E9708B6D1E52C8250029E0A5 /* gzip.c in Sources */,
 				10756E2C1AC126420009BF57 /* emitter.c in Sources */,
 				E9708B3D1E52C7860029E0A5 /* recv.c in Sources */,
+				0812174E1E07B89600712F36 /* redis.c in Sources */,
 				E9708B691E52C80F0029E0A5 /* http2_debug_state.c in Sources */,
 				100A55151C30C5BC00C4E3E0 /* chunked.c in Sources */,
 				E9708B3E1E52C7860029E0A5 /* send.c in Sources */,

diff --git a/include/h2o/mruby_.h b/include/h2o/mruby_.h
@@ -55,6 +55,7 @@ enum {
     H2O_MRUBY_PROC_EACH_TO_ARRAY,
     H2O_MRUBY_PROC_APP_TO_FIBER,
 
+    H2O_MRUBY_H2O_MODULE,
     H2O_MRUBY_GENERATOR_CLASS,
 
     /* used by chunked.c */
@@ -184,6 +185,9 @@ h2o_mruby_http_request_context_t *h2o_mruby_http_set_shortcut(mrb_state *mrb, mr
 void h2o_mruby_http_unset_shortcut(mrb_state *mrb, h2o_mruby_http_request_context_t *ctx, h2o_mruby_generator_t *generator);
 h2o_buffer_t **h2o_mruby_http_peek_content(h2o_mruby_http_request_context_t *ctx, int *is_final);
 
+/* handler/mruby/redis.c */
+void h2o_mruby_redis_init_context(h2o_mruby_shared_context_t *ctx);
+
 /* handler/mruby/sleep.c */
 void h2o_mruby_sleep_init_context(h2o_mruby_shared_context_t *ctx);
 

diff --git a/include/h2o/redis.h b/include/h2o/redis.h
@@ -27,37 +27,61 @@
 struct redisAsyncContext;
 struct redisReply;
 
+extern const char *const h2o_redis_error_connection;
+extern const char *const h2o_redis_error_protocol;
+extern const char *const h2o_redis_error_connect_timeout;
+extern const char *const h2o_redis_error_command_timeout;
+
 typedef enum {
     H2O_REDIS_CONNECTION_STATE_CLOSED = 0,
     H2O_REDIS_CONNECTION_STATE_CONNECTING,
     H2O_REDIS_CONNECTION_STATE_CONNECTED,
 } h2o_redis_connection_state_t;
 
-typedef struct st_h2o_redis_conn_t {
+typedef struct st_h2o_redis_client_t {
     h2o_loop_t *loop;
     h2o_redis_connection_state_t state;
     void (*on_connect)(void);
     void (*on_close)(const char *errstr);
+    uint64_t connect_timeout;
+    uint64_t command_timeout;
 
     struct redisAsyncContext *_redis;
     h2o_timeout_t _defer_timeout;
-    h2o_timeout_entry_t _timeout_entry;
-} h2o_redis_conn_t;
+    h2o_timeout_entry_t _defer_timeout_entry;
+    h2o_timeout_t _connect_timeout;
+    h2o_timeout_entry_t _connect_timeout_entry;
+} h2o_redis_client_t;
+
+typedef void (*h2o_redis_command_cb)(struct redisReply *reply, void *cb_data, const char *errstr);
 
-typedef void (*h2o_redis_command_cb)(struct redisReply *reply, void *cb_data);
+typedef enum enum_h2o_redis_command_type_t {
+    H2O_REDIS_COMMAND_TYPE_NORMAL = 1,
+    H2O_REDIS_COMMAND_TYPE_SUBSCRIBE,
+    H2O_REDIS_COMMAND_TYPE_UNSUBSCRIBE,
+    H2O_REDIS_COMMAND_TYPE_PSUBSCRIBE,
+    H2O_REDIS_COMMAND_TYPE_PUNSUBSCRIBE,
+    H2O_REDIS_COMMAND_TYPE_MONITOR,
+    H2O_REDIS_COMMAND_TYPE_ERROR
+} h2o_redis_command_type_t;
 
 typedef struct st_h2o_redis_command_t {
-    h2o_redis_conn_t *conn;
+    h2o_redis_client_t *client;
     h2o_redis_command_cb cb;
     void *data;
-    h2o_timeout_entry_t _timeout_entry;
+    h2o_redis_command_type_t type;
+    h2o_timeout_entry_t _defer_timeout_entry;
+    h2o_timeout_t _command_timeout;
+    h2o_timeout_entry_t _command_timeout_entry;
 } h2o_redis_command_t;
 
-h2o_redis_conn_t *h2o_redis_create_connection(h2o_loop_t *loop, size_t sz);
-void h2o_redis_connect(h2o_redis_conn_t *conn, const char *host, uint16_t port);
-void h2o_redis_disconnect(h2o_redis_conn_t *conn);
-void h2o_redis_free(h2o_redis_conn_t *conn);
+h2o_redis_client_t *h2o_redis_create_client(h2o_loop_t *loop, size_t sz);
+void h2o_redis_connect(h2o_redis_client_t *client, const char *host, uint16_t port);
+void h2o_redis_disconnect(h2o_redis_client_t *client);
+void h2o_redis_free(h2o_redis_client_t *client);
 
-h2o_redis_command_t *h2o_redis_command(h2o_redis_conn_t *conn, h2o_redis_command_cb cb, void *cb_data, const char *format, ...);
+h2o_redis_command_t *h2o_redis_command(h2o_redis_client_t *client, h2o_redis_command_cb cb, void *cb_data, const char *format, ...);
+h2o_redis_command_t *h2o_redis_command_argv(h2o_redis_client_t *client, h2o_redis_command_cb cb, void *cb_data, int argc,
+                                            const char **argv, const size_t *argvlen);
 
 #endif