for nginx-clojure#49 Access request BODY in rewrite handler

Author: xfeep

src/c/config

@@ -36,6 +36,7 @@ JNI_HEADER_2="${JNI_HEADER_1}/linux"
 
 CFLAGS="-I \"${JNI_HEADER_1}\" \
         -I \"${JNI_HEADER_2}\" \
+        -DNGX_CLOJURE_BE_SILENT_WITHOUT_JVM  \
         $CFLAGS"
 
 #for easy debug on linux or macosx  

src/c/ngx_http_clojure_mem.h

@@ -25,12 +25,12 @@
 #include <inttypes.h>
 #endif
 
-#define nginx_clojure_ver  2006 /*0.2.6*/
+#define nginx_clojure_ver  2007 /*0.2.7*/
 
 /*the least jar version required*/
 #define nginx_clojure_required_rt_lver 2006
 
-#define NGINX_CLOJURE_VER "nginx clojure/0.2.6"
+#define NGINX_CLOJURE_VER "nginx clojure/0.2.7"
 
 typedef struct {
 	ngx_int_t phrase;
@@ -40,14 +40,18 @@ typedef struct {
 	/*these two members only used by hijack send & read, default is 0*/
 	unsigned last_buf_meeted : 1;
 	unsigned ignore_filters : 1;
+	unsigned async_body_read : 1;
+	unsigned client_body_done : 1;
 } ngx_http_clojure_module_ctx_t;
 
 #define ngx_http_clojure_init_ctx(ctx, p) \
 		ctx->handled_couter = 1; \
 		ctx->phrase = p; \
 		ctx->last_buf_meeted = 0; \
 		ctx->busy = ctx->free = NULL; \
-		ctx->ignore_filters = 0
+		ctx->ignore_filters = 0; \
+		ctx->client_body_done = 0; \
+		ctx->async_body_read = 0
 
 #define NGX_HTTP_CLOJURE_MEM_IDX_START 0
 

src/c/ngx_http_clojure_module.c

@@ -656,8 +656,41 @@ static ngx_int_t   ngx_http_clojure_postconfiguration(ngx_conf_t *cf) {
 
 static void ngx_http_clojure_client_body_handler(ngx_http_request_t *r) {
 	ngx_http_clojure_loc_conf_t  *lcf = ngx_http_get_module_loc_conf(r, ngx_http_clojure_module);
-	int rc = ngx_http_clojure_eval(lcf->handler_id, r);
-	ngx_http_finalize_request (r , rc);
+	ngx_http_clojure_module_ctx_t *ctx = ngx_http_get_module_ctx(r, ngx_http_clojure_module);
+	ngx_int_t handler_id;
+	int rc = NGX_DECLINED;
+	int rewrite_phase = 0;
+	ctx->client_body_done = 1;
+
+	if (!ctx->async_body_read) {
+		ctx->phrase = -1;
+		return;
+	}
+
+	ctx->async_body_read = 0;
+
+	if (ctx->phrase == NGX_HTTP_REWRITE_PHASE) {
+		handler_id = lcf->rewrite_handler_id;
+		rewrite_phase = 1;
+		r->write_event_handler = ngx_http_core_run_phases;
+	}else {
+		handler_id = lcf->handler_id;
+	}
+
+	if (handler_id > 0)  {
+		rc = ngx_http_clojure_eval(handler_id, r);
+		ngx_log_debug2(NGX_LOG_DEBUG_HTTP, ngx_http_clojure_global_cycle->log, 0, "ngx clojure rewrite (body done callback) request: %" PRIu64 ", rc: %d", (jlong)(uintptr_t)r, rc);
+	}
+
+	if (rewrite_phase) {
+		r->main->count --;
+		if (rc != NGX_DONE) {
+			ctx->phrase = ~ctx->phrase;
+			r->write_event_handler(r);
+		}
+	}else {
+		ngx_http_finalize_request (r , rc);
+	}
 }
 
 
@@ -706,21 +739,24 @@ static ngx_int_t ngx_http_clojure_handler(ngx_http_request_t * r) {
 		r->request_body_in_single_buf = 1;
 		r->request_body_in_clean_file = 1;
 		r->request_body_in_persistent_file = 1;
+
 		rc = ngx_http_read_client_request_body(r, ngx_http_clojure_client_body_handler);
-    	if (rc >= NGX_HTTP_SPECIAL_RESPONSE) {
+    	if (rc == NGX_ERROR || rc >= NGX_HTTP_SPECIAL_RESPONSE) {
     		return rc;
     	}
-    	return NGX_DONE;
+
+    	if (rc == NGX_AGAIN) {
+    		ctx->async_body_read  = 1;
+    		return NGX_DONE;
+    	}
     }else {
     	rc = ngx_http_discard_request_body(r);
     	if (rc != NGX_OK && rc != NGX_AGAIN) {
     	   return rc;
     	}
-    	rc = ngx_http_clojure_eval(lcf->handler_id, r);
     }
-    //for debug
-    //ngx_log_error(NGX_LOG_DEBUG, r->connection->log, 0, "finished one request \n==============================================================\n\n");
 
+    rc = ngx_http_clojure_eval(lcf->handler_id, r);
     return rc;
 }
 
@@ -731,13 +767,43 @@ static ngx_int_t ngx_http_clojure_rewrite_handler(ngx_http_request_t * r) {
 	ngx_http_clojure_module_ctx_t *ctx;
 	ngx_http_clojure_loc_conf_t  *lcf = ngx_http_get_module_loc_conf(r, ngx_http_clojure_module);
 
-	if (!lcf->enable || (lcf->rewrite_handler_code.len == 0 && lcf->rewrite_handler_name.len == 0)) {
-		return NGX_DECLINED;
+	if (lcf->enable && (lcf->rewrite_handler_code.len > 0 || lcf->rewrite_handler_name.len > 0)) {
+		rc = ngx_http_clojure_init_clojure_script("rewrite handler", &lcf->handler_type, &lcf->rewrite_handler_name,
+				&lcf->rewrite_handler_code, &lcf->rewrite_handler_id, ngx_http_clojure_global_cycle->log);
+		if (rc != NGX_HTTP_CLOJURE_JVM_OK) {
+			return rc;
+		}
 	}
 
-	rc = ngx_http_clojure_init_clojure_script("rewrite handler", &lcf->handler_type, &lcf->rewrite_handler_name, &lcf->rewrite_handler_code, &lcf->rewrite_handler_id, ngx_http_clojure_global_cycle->log);
-	if (rc != NGX_HTTP_CLOJURE_JVM_OK) {
-			return rc;
+	if (lcf->always_read_body) {
+		if ((ctx = ngx_http_get_module_ctx(r, ngx_http_clojure_module)) == NULL) {
+			ctx = ngx_palloc(r->pool, sizeof(ngx_http_clojure_module_ctx_t));
+			if (ctx == NULL) {
+				ngx_log_error(NGX_LOG_ERR, r->connection->log, 0, "OutOfMemory of create ngx_http_clojure_module_ctx_t");
+				return NGX_HTTP_INTERNAL_SERVER_ERROR;
+			}
+
+			ngx_http_clojure_init_ctx(ctx, NGX_HTTP_REWRITE_PHASE);
+			ngx_http_set_ctx(r, ctx, ngx_http_clojure_module);
+
+			if (!ctx->client_body_done) {
+				r->request_body_in_single_buf = 1;
+				r->request_body_in_clean_file = 1;
+				r->request_body_in_persistent_file = 1;
+				rc = ngx_http_read_client_request_body(r, ngx_http_clojure_client_body_handler);
+		    	if (rc >= NGX_HTTP_SPECIAL_RESPONSE || rc == NGX_ERROR) {
+		    		return rc;
+		    	}
+		    	if (rc == NGX_AGAIN) {
+			    	ctx->async_body_read = 1;
+			    	return NGX_DONE;
+		    	}
+			}
+		}
+	}
+
+	if (!lcf->enable || (lcf->rewrite_handler_code.len == 0 && lcf->rewrite_handler_name.len == 0)) {
+		return NGX_DECLINED;
 	}
 
 	if ((ctx = ngx_http_get_module_ctx(r, ngx_http_clojure_module)) == NULL) {

test/clojure/nginx/clojure/asyn_channel_handlers_for_test.clj

@@ -44,7 +44,7 @@
                         (ashutdown! upstream :soft-write)
                         (arecv! upstream buf pipe error-handler pipe-handler)
                         )]
-    (aconnect! upstream "mirror.bit.edu.cn:80" nil error-handler
+    (aconnect! upstream "mirror.bit.edu.cn:80" upstream error-handler
                (fn [status att]
                  (.info logger "connected successfully")
                  (asend! upstream (str "GET /apache/httpcomponents/httpclient/RELEASE_NOTES-4.3.x.txt HTTP/1.1\r\n"

test/clojure/nginx/clojure/test_all.clj

@@ -556,6 +556,22 @@
              (debug-println "=================rewritesimple=============================")
              (is (= 200 (:status r)))
              (is (or (= "hello,a!/mytestpath" (:body r)) (= "hello,b!/mytestpath" (:body r))))))
+       (testing "rewrite proxy pass by body -- small body"
+           (let [r (client/post (str "http://" *host* ":" *port* "/javarewritebybodyproxy/") {:follow-redirects false, :body "ub"})
+                 h (:headers r)
+                 b (r :body)]
+             (debug-println r)
+             (debug-println "=================rewrite proxy pass by body -- small body=============================")
+             (is (= 200 (:status r)))
+             (is  (= "hello,b!/javarewritebybodyproxy/" (:body r)) )))
+        (testing "rewrite proxy pass by body -- large body"
+           (let [r (client/post (str "http://" *host* ":" *port* "/javarewritebybodyproxy/") {:follow-redirects false, :body (clojure.java.io/file "test/nginx-working-dir/post-rewrite-large-body-data")})
+                 h (:headers r)
+                 b (r :body)]
+             (debug-println r)
+             (debug-println "=================rewritesimple=============================")
+             (is (= 200 (:status r)))
+             (is  (= "hello,b!/javarewritebybodyproxy/" (:body r)) )))
   )
 
 ;eg. (concurrent-run 10 (run-tests 'nginx.clojure.test-all))

test/java/nginx/clojure/java/RewriteHandlerTestSet4NginxJavaRingHandler.java

@@ -1,5 +1,9 @@
 package nginx.clojure.java;
 
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
 import java.util.Map;
 
 import nginx.clojure.NginxClojureRT;
@@ -27,4 +31,32 @@ public Object[] invoke(Map<String, Object> request) {
 		}
 	}
 
+	
+	public static class SimpleRewriteByBodyHandler   implements NginxJavaRingHandler {
+		@Override
+		public Object[] invoke(Map<String, Object> request) {
+			NginxJavaRequest req = (NginxJavaRequest) request;
+			InputStream in = (InputStream) req.get(Constants.BODY);
+			if (in != null) {
+				long nr = req.nativeRequest();
+				BufferedReader reader;
+				try {
+					reader = new BufferedReader( new InputStreamReader(in, "utf-8"));
+					//just for test no arguments check
+					//read last line for test small or large body
+					String up = null, l = null;
+					while ( (l = reader.readLine()) != null) {
+						up = l;
+					}
+					System.err.println("[SimpleRewriteByBodyHandler] up=" + up + ", body inputstream type = " + in.getClass());
+					NginxClojureRT.setNGXVariable(nr, "myup", up);
+				} catch (IOException e) {
+					NginxClojureRT.UNSAFE.throwException(e);
+				}
+			}
+			return Constants.PHRASE_DONE;
+		}
+	}
+	
+	
 }

test/nginx-working-dir/conf/nginx-coroutine.conf

@@ -186,7 +186,6 @@ http {
           handler_type 'clojure';
           handler_code ' 
 						(fn[req]
-						  (println )
 						  {
 						    :status 200,
 						    :headers {"content-type" "text/plain"},
@@ -282,6 +281,13 @@ http {
           handler_name 'nginx.clojure.java.RewriteHandlerTestSet4NginxJavaRingHandler$SimpleVarHandler';
        }  
 
+       location /javarewritebybodyproxy {
+          always_read_body on;
+          handler_type 'java';
+          rewrite_handler_name 'nginx.clojure.java.RewriteHandlerTestSet4NginxJavaRingHandler$SimpleRewriteByBodyHandler';
+          proxy_pass http://$myup;
+       }
+       
        set $myup "";
        set $mypath "";
        location /uptest {

test/nginx-working-dir/conf/nginx-plain.conf

@@ -45,7 +45,7 @@ http {
     #gzip  on;
 
 
-     jvm_path "/usr/lib/jvm/java-7-oracle/jre/lib/amd64/server/libjvm.so";
+    jvm_path "/usr/lib/jvm/java-7-oracle/jre/lib/amd64/server/libjvm.so";
     #jvm_path "/usr/lib/jvm/jdk1.8.0_11/jre/lib/amd64/server/libjvm.so";
 
 
@@ -283,6 +283,13 @@ http {
           handler_name 'nginx.clojure.java.RewriteHandlerTestSet4NginxJavaRingHandler$SimpleVarHandler';
        }  
 
+       location /javarewritebybodyproxy {
+          always_read_body on;
+          handler_type 'java';
+          rewrite_handler_name 'nginx.clojure.java.RewriteHandlerTestSet4NginxJavaRingHandler$SimpleRewriteByBodyHandler';
+          proxy_pass http://$myup;
+       }
+       
        set $myup "";
        set $mypath "";
        location /uptest {

test/nginx-working-dir/conf/nginx-threadpool.conf

@@ -186,7 +186,6 @@ http {
           handler_type 'clojure';
           handler_code ' 
 						(fn[req]
-						  (println )
 						  {
 						    :status 200,
 						    :headers {"content-type" "text/plain"},
@@ -282,6 +281,13 @@ http {
           handler_name 'nginx.clojure.java.RewriteHandlerTestSet4NginxJavaRingHandler$SimpleVarHandler';
        }  
 
+       location /javarewritebybodyproxy {
+          always_read_body on;
+          handler_type 'java';
+          rewrite_handler_name 'nginx.clojure.java.RewriteHandlerTestSet4NginxJavaRingHandler$SimpleRewriteByBodyHandler';
+          proxy_pass http://$myup;
+       }
+       
        set $myup "";
        set $mypath "";
        location /uptest {