增加 Message.end 用于中止 chain 事务

fibjs · Mar 11, 2017 · b61b9d9 · b61b9d9
1 parent 232657c
commit b61b9d9
Show file tree

Hide file tree

Showing 12 changed files with 129 additions and 12 deletions.
diff --git a/fibjs/include/HttpRequest.h b/fibjs/include/HttpRequest.h
@@ -40,6 +40,8 @@ class HttpRequest: public HttpRequest_base
     virtual result_t readAll(obj_ptr<Buffer_base> &retVal, AsyncEvent *ac);
     virtual result_t write(Buffer_base *data, AsyncEvent *ac);
     virtual result_t get_length(int64_t &retVal);
+    virtual result_t end();
+    virtual result_t isEnded(bool& retVal);
     virtual result_t clear();
     virtual result_t sendTo(Stream_base *stm, AsyncEvent *ac);
     virtual result_t readFrom(Stream_base *stm, AsyncEvent *ac);

diff --git a/fibjs/include/HttpResponse.h b/fibjs/include/HttpResponse.h
@@ -39,6 +39,8 @@ class HttpResponse: public HttpResponse_base
     virtual result_t readAll(obj_ptr<Buffer_base> &retVal, AsyncEvent *ac);
     virtual result_t write(Buffer_base *data, AsyncEvent *ac);
     virtual result_t get_length(int64_t &retVal);
+    virtual result_t end();
+    virtual result_t isEnded(bool& retVal);
     virtual result_t clear();
     virtual result_t sendTo(Stream_base *stm, AsyncEvent *ac);
     virtual result_t readFrom(Stream_base *stm, AsyncEvent *ac);

diff --git a/fibjs/include/Message.h b/fibjs/include/Message.h
@@ -31,6 +31,8 @@ class Message: public Message_base
     virtual result_t readAll(obj_ptr<Buffer_base> &retVal, AsyncEvent *ac);
     virtual result_t write(Buffer_base *data, AsyncEvent *ac);
     virtual result_t get_length(int64_t &retVal);
+    virtual result_t end();
+    virtual result_t isEnded(bool& retVal);
     virtual result_t clear();
     virtual result_t sendTo(Stream_base *stm, AsyncEvent *ac);
     virtual result_t readFrom(Stream_base *stm, AsyncEvent *ac);
@@ -40,7 +42,7 @@ class Message: public Message_base
     virtual result_t set_lastError(exlib::string newVal);
 
 public:
-    Message(bool bRep = false) : m_bRep(bRep)
+    Message(bool bRep = false) : m_bRep(bRep), m_end(false)
     {
     }
 
@@ -60,6 +62,7 @@ class Message: public Message_base
     obj_ptr<SeekableStream_base> m_body;
     exlib::string m_lastError;
     bool m_bRep;
+    bool m_end;
 };
 
 } /* namespace fibjs */

diff --git a/fibjs/include/WebSocketMessage.h b/fibjs/include/WebSocketMessage.h
@@ -37,6 +37,8 @@ class WebSocketMessage: public WebSocketMessage_base
     virtual result_t readAll(obj_ptr<Buffer_base> &retVal, AsyncEvent *ac);
     virtual result_t write(Buffer_base *data, AsyncEvent *ac);
     virtual result_t get_length(int64_t &retVal);
+    virtual result_t end();
+    virtual result_t isEnded(bool& retVal);
     virtual result_t clear();
     virtual result_t sendTo(Stream_base *stm, AsyncEvent *ac);
     virtual result_t readFrom(Stream_base *stm, AsyncEvent *ac);

diff --git a/fibjs/include/ifs/Message.h b/fibjs/include/ifs/Message.h
@@ -41,6 +41,8 @@ class Message_base : public object_base
     virtual result_t readAll(obj_ptr<Buffer_base>& retVal, AsyncEvent* ac) = 0;
     virtual result_t write(Buffer_base* data, AsyncEvent* ac) = 0;
     virtual result_t get_length(int64_t& retVal) = 0;
+    virtual result_t end() = 0;
+    virtual result_t isEnded(bool& retVal) = 0;
     virtual result_t clear() = 0;
     virtual result_t sendTo(Stream_base* stm, AsyncEvent* ac) = 0;
     virtual result_t readFrom(Stream_base* stm, AsyncEvent* ac) = 0;
@@ -67,6 +69,8 @@ class Message_base : public object_base
     static void s_readAll(const v8::FunctionCallbackInfo<v8::Value>& args);
     static void s_write(const v8::FunctionCallbackInfo<v8::Value>& args);
     static void s_get_length(v8::Local<v8::String> property, const v8::PropertyCallbackInfo<v8::Value> &args);
+    static void s_end(const v8::FunctionCallbackInfo<v8::Value>& args);
+    static void s_isEnded(const v8::FunctionCallbackInfo<v8::Value>& args);
     static void s_clear(const v8::FunctionCallbackInfo<v8::Value>& args);
     static void s_sendTo(const v8::FunctionCallbackInfo<v8::Value>& args);
     static void s_readFrom(const v8::FunctionCallbackInfo<v8::Value>& args);
@@ -99,6 +103,8 @@ namespace fibjs
             {"read", s_read, false},
             {"readAll", s_readAll, false},
             {"write", s_write, false},
+            {"end", s_end, false},
+            {"isEnded", s_isEnded, false},
             {"clear", s_clear, false},
             {"sendTo", s_sendTo, false},
             {"readFrom", s_readFrom, false}
@@ -306,6 +312,32 @@ namespace fibjs
         METHOD_RETURN();
     }
 
+    inline void Message_base::s_end(const v8::FunctionCallbackInfo<v8::Value>& args)
+    {
+        METHOD_INSTANCE(Message_base);
+        METHOD_ENTER();
+
+        METHOD_OVER(0, 0);
+
+        hr = pInst->end();
+
+        METHOD_VOID();
+    }
+
+    inline void Message_base::s_isEnded(const v8::FunctionCallbackInfo<v8::Value>& args)
+    {
+        bool vr;
+
+        METHOD_INSTANCE(Message_base);
+        METHOD_ENTER();
+
+        METHOD_OVER(0, 0);
+
+        hr = pInst->isEnded(vr);
+
+        METHOD_RETURN();
+    }
+
     inline void Message_base::s_clear(const v8::FunctionCallbackInfo<v8::Value>& args)
     {
         METHOD_INSTANCE(Message_base);

diff --git a/fibjs/include/ifs/Message.idl b/fibjs/include/ifs/Message.idl
@@ -43,6 +43,14 @@ interface Message : object
     /*! @brief 消息数据部分的长度 */
     readonly Long length;
 
+    /*! @brief 设置当前消息处理结束，Chain 处理器不再继续后面的事务 */
+    end();
+
+    /*! @brief 查询当前消息是否结束
+     @return 结束则返回 true
+    */
+    Boolean isEnded();
+
     /*! @brief 清除消息的内容 */
     clear();
 

diff --git a/fibjs/src/http/HttpRequest.cpp b/fibjs/src/http/HttpRequest.cpp
@@ -185,6 +185,16 @@ result_t HttpRequest::set_lastError(exlib::string newVal)
     return m_message->set_lastError(newVal);
 }
 
+result_t HttpRequest::end()
+{
+    return m_message->end();
+}
+
+result_t HttpRequest::isEnded(bool& retVal)
+{
+    return m_message->isEnded(retVal);
+}
+
 result_t HttpRequest::clear()
 {
     m_message = new HttpMessage();

diff --git a/fibjs/src/http/HttpResponse.cpp b/fibjs/src/http/HttpResponse.cpp
@@ -186,6 +186,16 @@ result_t HttpResponse::set_lastError(exlib::string newVal)
     return m_message->set_lastError(newVal);
 }
 
+result_t HttpResponse::end()
+{
+    return m_message->end();
+}
+
+result_t HttpResponse::isEnded(bool& retVal)
+{
+    return m_message->isEnded(retVal);
+}
+
 result_t HttpResponse::clear()
 {
     m_message = new HttpMessage(true);

diff --git a/fibjs/src/mq/Chain.cpp b/fibjs/src/mq/Chain.cpp
@@ -40,6 +40,8 @@ result_t Chain::invoke(object_base *v, obj_ptr<Handler_base> &retVal,
         {
             int32_t i;
 
+            m_msg = Message_base::getInstance(m_v);
+
             m_array.resize(pThis->m_array.size());
             for (i = 0; i < (int32_t) pThis->m_array.size(); i ++)
                 m_array[i] = pThis->m_array[i];
@@ -51,8 +53,11 @@ result_t Chain::invoke(object_base *v, obj_ptr<Handler_base> &retVal,
         static int32_t invoke(AsyncState *pState, int32_t n)
         {
             asyncInvoke *pThis = (asyncInvoke *) pState;
+            bool end = false;
 
-            if (pThis->m_pos == (int32_t) pThis->m_array.size())
+            if (pThis->m_msg)
+                pThis->m_msg->isEnded(end);
+            if (end || (pThis->m_pos == (int32_t) pThis->m_array.size()))
                 return pThis->done(CALL_RETURN_NULL);
 
             pThis->m_pos++;
@@ -63,6 +68,7 @@ result_t Chain::invoke(object_base *v, obj_ptr<Handler_base> &retVal,
     private:
         std::vector<obj_ptr<Handler_base> > m_array;
         obj_ptr<object_base> m_v;
+        obj_ptr<Message_base> m_msg;
         int32_t m_pos;
     };
 

diff --git a/fibjs/src/mq/Message.cpp b/fibjs/src/mq/Message.cpp
@@ -111,8 +111,24 @@ result_t Message_base::_new(obj_ptr<Message_base> &retVal, v8::Local<v8::Object>
     return 0;
 }
 
+result_t Message::end()
+{
+    m_end = true;
+    return 0;
+}
+
+result_t Message::isEnded(bool& retVal)
+{
+    retVal = m_end;
+    if (!m_end && m_response)
+        m_response->isEnded(retVal);
+
+    return 0;
+}
+
 result_t Message::clear()
 {
+    m_end = false;
     m_params.Release();
     m_result.clear();
     m_value.clear();

diff --git a/fibjs/src/websocket/WebSocketMessage.cpp b/fibjs/src/websocket/WebSocketMessage.cpp
@@ -93,6 +93,16 @@ result_t WebSocketMessage::set_lastError(exlib::string newVal)
     return m_message->set_lastError(newVal);
 }
 
+result_t WebSocketMessage::end()
+{
+    return m_message->end();
+}
+
+result_t WebSocketMessage::isEnded(bool& retVal)
+{
+    return m_message->isEnded(retVal);
+}
+
 result_t WebSocketMessage::clear()
 {
     m_message = new Message(m_bRep);

diff --git a/test/mq_test.js b/test/mq_test.js
@@ -135,16 +135,15 @@ describe("mq", () => {
     })
 
     describe("chain handler", () => {
-        it("chain invoke",
-            () => {
-                var chain = new mq.Chain([hdlr1, hdlr2,
-                    mq.jsHandler(hdlr3)
-                ]);
+        it("chain invoke", () => {
+            var chain = new mq.Chain([hdlr1, hdlr2,
+                mq.jsHandler(hdlr3)
+            ]);
 
-                n = 0;
-                chain.invoke(v);
-                assert.equal(7, n);
-            });
+            n = 0;
+            chain.invoke(v);
+            assert.equal(7, n);
+        });
 
         it("params", () => {
             function chain_params(v, p1, p2) {
@@ -169,7 +168,6 @@ describe("mq", () => {
 
         it("Message", () => {
             var handler = new mq.Chain([
-
                 (v) => {
                     return {};
                 },
@@ -198,6 +196,24 @@ describe("mq", () => {
             mq.invoke(handler, req);
         });
 
+        it("end chain", () => {
+            var handler = new mq.Chain([
+                (v) => {
+                    return 1;
+                },
+                (v) => {
+                    v.end();
+                },
+                (v) => {
+                    return 3;
+                }
+            ]);
+
+            var req = new mq.Message();
+            mq.invoke(handler, req);
+            assert.equal(1, req.result);
+        });
+
         it("memory leak", () => {
             var svr = new net.TcpServer(8888, () => {});
             ss.push(svr.socket);