Support user interceptor of server

src/brpc/interceptor.h

@@ -0,0 +1,42 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#ifndef BRPC_INTERCEPTOR_H
+#define BRPC_INTERCEPTOR_H
+
+#include "brpc/controller.h"
+
+
+namespace brpc {
+
+class Interceptor {
+public:
+    virtual ~Interceptor() = default;
+
+    // Returns true if accept request, reject request otherwise.
+    // When server rejects request, You can fill `error_code'
+    // and `error_txt' which will send to client.
+    virtual bool Accept(const brpc::Controller* controller,
+                        int& error_code,
+                        std::string& error_txt) const = 0;
+
+};
+
+} // namespace brpc
+
+
+#endif //BRPC_INTERCEPTOR_H

src/brpc/policy/baidu_rpc_protocol.cpp

@@ -309,6 +309,22 @@ void EndRunningCallMethodInPool(
     return EndRunningUserCodeInPool(CallMethodInBackupThread, args);
 };
 
+// Returns true if accept request, reject request otherwise.
+bool AcceptRequest(const Server* server, Controller* cntl) {
+    const Interceptor* interceptor = server->options().interceptor;
+    int error_code = 0;
+    std::string error_text;
+    if (interceptor &&
+        !interceptor->Accept(cntl, error_code, error_text)) {
+        cntl->SetFailed(error_code,
+                        "Reject by Interceptor: %s",
+                        error_text.c_str());
+        return false;
+    }
+
+    return true;
+}
+
 void ProcessRpcRequest(InputMessageBase* msg_base) {
     const int64_t start_parse_us = butil::cpuwide_time_us();
     DestroyingPtr<MostCommonMessage> msg(static_cast<MostCommonMessage*>(msg_base));
@@ -462,6 +478,11 @@ void ProcessRpcRequest(InputMessageBase* msg_base) {
         google::protobuf::Service* svc = mp->service;
         const google::protobuf::MethodDescriptor* method = mp->method;
         accessor.set_method(method);
+
+        if (!AcceptRequest(server, cntl.get())) {
+            break;
+        }
+
         if (span) {
             span->ResetServerSpanName(method->full_name());
         }

src/brpc/policy/http_rpc_protocol.cpp

@@ -1251,6 +1251,9 @@ void EndRunningCallMethodInPool(
     ::google::protobuf::Message* response,
     ::google::protobuf::Closure* done);
 
+// Defined in baidu_rpc_protocol.cpp
+bool AcceptRequest(const Server* server, Controller* cntl);
+
 void ProcessHttpRequest(InputMessageBase *msg) {
     const int64_t start_parse_us = butil::cpuwide_time_us();
     DestroyingPtr<HttpContext> imsg_guard(static_cast<HttpContext*>(msg));
@@ -1430,6 +1433,9 @@ void ProcessHttpRequest(InputMessageBase *msg) {
                             " -usercode_in_pthread is on");
             return;
         }
+        if (!AcceptRequest(server, cntl)) {
+            return;
+        }
     } else if (security_mode) {
         cntl->SetFailed(EPERM, "Not allowed to access builtin services, try "
                         "ServerOptions.internal_port=%d instead if you're in"

src/brpc/policy/hulu_pbrpc_protocol.cpp

@@ -328,6 +328,9 @@ void EndRunningCallMethodInPool(
     ::google::protobuf::Message* response,
     ::google::protobuf::Closure* done);
 
+// Defined in baidu_rpc_protocol.cpp
+bool AcceptRequest(const Server* server, Controller* cntl);
+
 void ProcessHuluRequest(InputMessageBase* msg_base) {
     const int64_t start_parse_us = butil::cpuwide_time_us();
     DestroyingPtr<MostCommonMessage> msg(static_cast<MostCommonMessage*>(msg_base));
@@ -469,6 +472,11 @@ void ProcessHuluRequest(InputMessageBase* msg_base) {
         google::protobuf::Service* svc = sp->service;
         const google::protobuf::MethodDescriptor* method = sp->method;
         accessor.set_method(method);
+
+        if (!AcceptRequest(server, cntl.get())) {
+            break;
+        }
+
         if (span) {
             span->ResetServerSpanName(method->full_name());
         }

src/brpc/policy/nshead_protocol.cpp

@@ -202,6 +202,9 @@ static void EndRunningCallMethodInPool(NsheadService* service,
     return EndRunningUserCodeInPool(CallMethodInBackupThread, args);
 };
 
+// Defined in baidu_rpc_protocol.cpp
+bool AcceptRequest(const Server* server, Controller* cntl);
+
 void ProcessNsheadRequest(InputMessageBase* msg_base) {
     const int64_t start_parse_us = butil::cpuwide_time_us();   
 
@@ -315,6 +318,9 @@ void ProcessNsheadRequest(InputMessageBase* msg_base) {
                             " -usercode_in_pthread is on");
             break;
         }
+        if (!AcceptRequest(server, cntl)) {
+            break;
+        }
     } while (false);
 
     msg.reset();  // optional, just release resource ASAP

src/brpc/policy/sofa_pbrpc_protocol.cpp

@@ -305,6 +305,9 @@ void EndRunningCallMethodInPool(
     ::google::protobuf::Message* response,
     ::google::protobuf::Closure* done);
 
+// Defined in baidu_rpc_protocol.cpp
+bool AcceptRequest(const Server* server, Controller* cntl);
+
 void ProcessSofaRequest(InputMessageBase* msg_base) {
     const int64_t start_parse_us = butil::cpuwide_time_us();
     DestroyingPtr<MostCommonMessage> msg(static_cast<MostCommonMessage*>(msg_base));
@@ -420,6 +423,11 @@ void ProcessSofaRequest(InputMessageBase* msg_base) {
         google::protobuf::Service* svc = sp->service;
         const google::protobuf::MethodDescriptor* method = sp->method;
         accessor.set_method(method);
+
+        if (!AcceptRequest(server, cntl.get())) {
+            break;
+        }
+
         if (span) {
             span->ResetServerSpanName(method->full_name());
         }

src/brpc/policy/thrift_protocol.cpp

@@ -446,6 +446,9 @@ static void EndRunningCallMethodInPool(ThriftService* service,
     return EndRunningUserCodeInPool(CallMethodInBackupThread, args);
 };
 
+// Defined in baidu_rpc_protocol.cpp
+bool AcceptRequest(const Server* server, Controller* cntl);
+
 void ProcessThriftRequest(InputMessageBase* msg_base) {
     const int64_t start_parse_us = butil::cpuwide_time_us();   
 
@@ -538,6 +541,10 @@ void ProcessThriftRequest(InputMessageBase* msg_base) {
                 " -usercode_in_pthread is on");
     }
 
+    if (!AcceptRequest(server, cntl)) {
+        return;
+    }
+
     msg.reset();  // optional, just release resource ASAP
 
     if (span) {

src/brpc/server.cpp

@@ -127,6 +127,8 @@ ServerOptions::ServerOptions()
     , mongo_service_adaptor(NULL)
     , auth(NULL)
     , server_owns_auth(false)
+    , interceptor(NULL)
+    , server_owns_interceptor(false)
     , num_threads(8)
     , max_concurrency(0)
     , session_local_data_factory(NULL)
@@ -451,6 +453,10 @@ Server::~Server() {
         delete _options.auth;
         _options.auth = NULL;
     }
+    if (_options.interceptor) {
+        delete _options.interceptor;
+        _options.interceptor = NULL;
+    }
 
     delete _options.redis_service;
     _options.redis_service = NULL;

src/brpc/server.h

@@ -41,6 +41,7 @@
 #include "brpc/adaptive_max_concurrency.h"
 #include "brpc/http2.h"
 #include "brpc/redis.h"
+#include "brpc/interceptor.h"
 
 namespace brpc {
 
@@ -91,6 +92,15 @@ struct ServerOptions {
     // Default: false
     bool server_owns_auth;
 
+    // Turn on request interception  if `interceptor' is not NULL.
+    // Default: NULL
+    const Interceptor* interceptor;
+
+    // false: `interceptor' is not owned by server and must be valid when server is running.
+    // true:  `interceptor' is owned by server and will be deleted when server is destructed.
+    // Default: false
+    bool server_owns_interceptor;
+
     // Number of pthreads that server runs on. Notice that this is just a hint,
     // you can't assume that the server uses exactly so many pthreads because
     // pthread workers are shared by all servers and channels inside a

test/brpc_interceptor_unittest.cpp

@@ -0,0 +1,157 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements.  See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership.  The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License.  You may obtain a copy of the License at
+//
+//   http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied.  See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+#include <gtest/gtest.h>
+#include <gflags/gflags.h>
+#include "brpc/policy/sofa_pbrpc_protocol.h"
+#include "brpc/channel.h"
+#include "brpc/server.h"
+#include "brpc/interceptor.h"
+#include "echo.pb.h"
+
+namespace brpc {
+namespace policy {
+DECLARE_bool(use_http_error_code);
+}
+}
+
+int main(int argc, char* argv[]) {
+    ::testing::InitGoogleTest(&argc, argv);
+    GFLAGS_NS::ParseCommandLineFlags(&argc, &argv, true);
+    return RUN_ALL_TESTS();
+}
+
+const int EREJECT = 4000;
+int g_index = 0;
+const int port = 8613;
+const std::string EXP_REQUEST = "hello";
+const std::string EXP_RESPONSE = "world";
+
+class EchoServiceImpl : public ::test::EchoService {
+public:
+    EchoServiceImpl() = default;
+    ~EchoServiceImpl() override = default;
+    void Echo(google::protobuf::RpcController* cntl_base,
+              const ::test::EchoRequest* request,
+              ::test::EchoResponse* response,
+              google::protobuf::Closure* done) override {
+        brpc::ClosureGuard done_guard(done);
+        EXPECT_EQ(EXP_REQUEST, request->message());
+        response->set_message(EXP_RESPONSE);
+    }
+};
+
+class MyInterceptor : public brpc::Interceptor {
+public:
+    MyInterceptor() = default;
+
+    ~MyInterceptor() override = default;
+
+    bool Accept(const brpc::Controller* controller,
+                int& error_code,
+                std::string& error_txt) const override {
+        if (g_index % 2 == 0) {
+            error_code = EREJECT;
+            error_txt = "reject g_index=0";
+            return false;
+        }
+
+        return true;
+    }
+};
+
+class InterceptorTest : public ::testing::Test {
+public:
+    InterceptorTest() {
+        EXPECT_EQ(0, _server.AddService(&_echo_svc,
+                                        brpc::SERVER_DOESNT_OWN_SERVICE));
+        brpc::ServerOptions options;
+        options.interceptor = new MyInterceptor;
+        options.server_owns_interceptor = true;
+        EXPECT_EQ(0, _server.Start(port, &options));
+    }
+
+    ~InterceptorTest() override = default;
+
+    void CallMethod(test::EchoService_Stub& stub,
+                    ::test::EchoRequest& req,
+                    ::test::EchoResponse& res) {
+        for (g_index = 0; g_index < 1000; ++g_index) {
+            brpc::Controller cntl;
+            stub.Echo(&cntl, &req, &res, NULL);
+            if (g_index % 2 == 0) {
+                ASSERT_TRUE(cntl.Failed());
+                ASSERT_EQ(EREJECT, cntl.ErrorCode());
+            } else {
+                ASSERT_FALSE(cntl.Failed());
+                EXPECT_EQ(EXP_RESPONSE, res.message()) << cntl.ErrorText();
+            }
+        }
+    }
+
+private:
+    brpc::Server _server;
+    EchoServiceImpl _echo_svc;
+};
+
+TEST_F(InterceptorTest, sanity) {
+    ::test::EchoRequest req;
+    ::test::EchoResponse res;
+    req.set_message(EXP_REQUEST);
+
+    // PROTOCOL_BAIDU_STD
+    {
+        brpc::Channel channel;
+        brpc::ChannelOptions options;
+        ASSERT_EQ(0, channel.Init("localhost", port, &options));
+        test::EchoService_Stub stub(&channel);
+        CallMethod(stub, req, res);
+    }
+
+    // PROTOCOL_HTTP
+    {
+        brpc::Channel channel;
+        brpc::ChannelOptions options;
+        options.protocol = brpc::PROTOCOL_HTTP;
+        ASSERT_EQ(0, channel.Init("localhost", port, &options));
+        test::EchoService_Stub stub(&channel);
+        // Set the x-bd-error-code header of http response to brpc error code.
+        brpc::policy::FLAGS_use_http_error_code = true;
+        CallMethod(stub, req, res);
+    }
+
+    // PROTOCOL_HULU_PBRPC
+    {
+        brpc::Channel channel;
+        brpc::ChannelOptions options;
+        options.protocol = brpc::PROTOCOL_HULU_PBRPC;
+        ASSERT_EQ(0, channel.Init("localhost", port, &options));
+        test::EchoService_Stub stub(&channel);
+        CallMethod(stub, req, res);
+    }
+
+    g_index = 0;
+    // PROTOCOL_SOFA_PBRPC
+    {
+        brpc::Channel channel;
+        brpc::ChannelOptions options;
+        options.protocol = brpc::PROTOCOL_SOFA_PBRPC;
+        ASSERT_EQ(0, channel.Init("localhost", port, &options));
+        test::EchoService_Stub stub(&channel);
+        CallMethod(stub, req, res);
+    }
+}