-
Notifications
You must be signed in to change notification settings - Fork 205
Hprose 中间件
Hprose 过滤器的功能虽然比较强大，可以将 Hprose 的功能进行扩展。但是有些功能使用它仍然难以实现，比如缓存。
为此，Hprose 2.0 引入了更加强大的中间件功能。Hprose 中间件不仅可以对输入输出的数据进行操作，它还可以对调用本身的参数和结果进行操作，甚至你可以跳过中间的执行步骤，或者完全由你来接管中间数据的处理。
Hprose 中间件跟普通的 HTTP 服务器中间件有些类似，但又有所不同。
Hprose 中间件在客户端服务器端都支持。
Hprose 中间件分为两种：
- 调用中间件
- 输入输出中间件
另外，输入输出中间件又可以细分为 BeforeFilter 和 AfterFilter 两种，但它们本质上没有什么区别，只是在执行顺序上有所区别。
Hprose 中间件的顺序执行是按照添加的前后顺序执行的，假设添加的中间件处理器分别为：handler1, handler2 ... handlerN，那么执行顺序就是 handler1, handler2 ... handlerN。
不同类型的 Hprose 中间件和 Hprose 其它过程的执行流程如下图所示：
+------------------------------------------------------------------+
| +-----------------batch invoke----------------+ |
| +------+ | +-----+ +------+ +------+ +-----+ | |
| |invoke| | |begin| |invoke| ... |invoke| | end | | |
| +------+ | +-----+ +------+ +------+ +-----+ | |
| ^ +---------------------------------------------+ |
| | ^ |
| | | |
| v v |
| +-------------------+ +------------------+ |
| | invoke middleware | | batch middleware | |
| +-------------------+ +------------------+ |
| ^ ^ |
| | +---------------+ | |
| +---->| encode/decode |<-----+ |
| +---------------+ |
| ^ |
| | |
| v |
| +--------------------------+ |
| | before filter middleware | |
| +--------------------------+ |
| ^ |
| | _ _ ___ ____ ____ ____ ____ |
| v |__| |__] |__/ | | [__ |___ |
| +--------+ | | | | \ |__| ___] |___ |
| | filter | |
| +--------+ ____ _ _ ____ _ _ ___ |
| ^ | | | |___ |\ | | |
| | |___ |___ | |___ | \| | |
| v |
| +-------------------------+ |
| | after filter middleware | |
| +-------------------------+ |
+------------------------------------------------------------------+
^
|
|
v
+------------------------------------------------------------------+
| +--------------------------+ |
| | before filter middleware | |
| +--------------------------+ |
| ^ |
| | _ _ ___ ____ ____ ____ ____ |
| v |__| |__] |__/ | | [__ |___ |
| +--------+ | | | | \ |__| ___] |___ |
| | filter | |
| +--------+ ____ ____ ____ _ _ ____ ____ |
| ^ [__ |___ |__/ | | |___ |__/ |
| | ___] |___ | \ \/ |___ | \ |
| v |
| +-------------------------+ |
| | after filter middleware | |
| +-------------------------+ |
| ^ |
| | |
| v |
| +---------------+ |
| +----------->| encode/decode |<---------------------+ |
| | +---------------+ | |
| | | | |
| | | | |
| | v | |
| | +---------------+ | |
| | | before invoke |-------------+ | |
| | +---------------+ | | |
| | | | | |
| | | | | |
| | v v | |
| | +-------------------+ +------------+ | |
| | | invoke middleware |--->| send error |--+ |
| | +-------------------+ +------------+ |
| | | ^ |
| | | | |
| | v | |
| | +--------------+ | |
| | | after invoke |--------------+ |
| | +--------------+ |
| | | |
| | | |
| +--------------------+ |
+------------------------------------------------------------------+
调用中间件的形式为：
func(
name string,
args []reflect.Value,
context Context,
next NextInvokeHandler) (results []reflect.Value, err error) {
...
results, err = next(name, args, context)
...
return results, err
}name 是调用的远程函数/方法名。
args 是调用参数，引用传递的数组。
context 是调用上下文对象。
next 表示下一个中间件。通过调用 next 将各个中间件串联起来。
在调用 next 之前的操作在调用发生前执行，在调用 next 之后的操作在调用发生后执行，如果你不想修改返回结果，你应该将 next 的返回值作为该中间件的返回值返回。
如果在调用 next 之前的操作中发生 panic，则 next 和后面的代码都不会被执行而是直接返回上一层，panic 信息作为 error 结果返回给上一层中间件。
调用中间件可以是普通函数/方法，也可以是匿名函数。
我们来看一个例子：
package main
import (
"fmt"
"reflect"
"github.com/hprose/hprose-golang/rpc"
)
func logInvokeHandler(
name string,
args []reflect.Value,
context rpc.Context,
next rpc.NextInvokeHandler) (results []reflect.Value, err error) {
fmt.Printf("%s(%v) = ", name, args)
results, err = next(name, args, context)
fmt.Printf("%v %v\r\n", results, err)
return
}
func hello(name string) string {
return "Hello " + name + "!"
}
type HelloService struct {
Hello func(string) (string, error)
Hi func(string) error
}
func main() {
server := rpc.NewTCPServer("")
server.AddInvokeHandler(logInvokeHandler)
server.AddFunction("hello", hello)
server.Handle()
client := rpc.NewClient(server.URI())
var helloService *HelloService
client.UseService(&helloService)
helloService.Hello("World")
helloService.Hi("World")
client.Close()
server.Close()
}运行该程序，我们会看到如下结果：
Hello([World]) = [Hello World!] <nil>
Hi([<interface {} Value>]) = [] Can't find this method Hi
如果把：
server.AddInvokeHandler(logInvokeHandler)去掉，在创建客户端之后，加上下面这句：
client.AddInvokeHandler(logInvokeHandler)再运行该程序，我们会看到如下输出：
Hello([World]) = [Hello World!] <nil>
Hi([World]) = [] Can't find this method Hi
我们会发现，关于 Hello 方法调用的输出是一样的，关于 Hi 方法的调用，参数上不太一样。原因是客户端的参数总是跟调用时带入的参数值类型一致的，而在服务器端的参数类型是跟服务器端服务函数/方法的参数定义类型一致，而 Hi 方法在服务器并没有定义，因此，在反序列化时，是按照 interface{} 类型反序列化的，所以在服务器端看到的是一个 interface{} 类型的值。
输入输出中间件可以完全代替 Hprose 过滤器。使用输入输出中间件还是使用 Hprose 过滤器完全看开发者喜好。
输入输出中间件的形式为：
func(
request []byte,
context Context,
next NextFilterHandler) (response []byte, err error) {
...
response, err = next(request, context)
...
return response, errrequest 是原始请求数据，对于客户端来说它是输出数据，对于服务器端来说，它是输入数据。
context 是调用上下文对象。
next 表示下一个中间件。通过调用 next 将各个中间件串联起来。
next 的返回值 response 是返回的响应数据。对于客户端来说，它是输入数据。对于服务器端来说，它是输出数据。
下面我们来看一下 Hprose 过滤器中的跟踪调试的例子在这里如何实现。
package main
import (
"fmt"
"github.com/hprose/hprose-golang/rpc"
)
type logFilter struct {
Prompt string
}
func (lf logFilter) handler(
request []byte,
context rpc.Context,
next rpc.NextFilterHandler) (response []byte, err error) {
fmt.Printf("%v: %s\r\n", lf.Prompt, request)
response, err = next(request, context)
fmt.Printf("%v: %s\r\n", lf.Prompt, response)
return
}
func hello(name string) string {
return "Hello " + name + "!"
}
type HelloService struct {
Hello func(string) (string, error)
}
func main() {
server := rpc.NewTCPServer("")
server.AddFunction("hello", hello)
server.AddBeforeFilterHandler(logFilter{"Server"}.handler)
server.Handle()
client := rpc.NewClient(server.URI())
client.AddBeforeFilterHandler(logFilter{"Client"}.handler)
var helloService *HelloService
client.UseService(&helloService)
helloService.Hello("World")
client.Close()
server.Close()
}执行结果为：
Client: Cs5"Hello"a1{s5"World"}z
Server: Cs5"Hello"a1{s5"World"}z
Server: Rs12"Hello World!"z
Client: Rs12"Hello World!"z
这个结果跟使用 Hprose 过滤器的例子的结果一样。
但是我们发现，这里使用 Hprose 中间件要写的代码比起 Hprose 过滤器更简单，只需要一个方法就可以了。
另外，因为这个例子中，我们没有使用过滤器功能，因此使用 AddBeforeFilterHandler（或 BeforeFilter.Use）方法或者 AddAfterFilterHandler（或 AfterFilter.Use）方法添加中间件处理器效果都是一样的。
但如果我们使用了过滤器的话，那么 AddBeforeFilterHandler 添加的中间件处理器的 request 数据是未经过过滤器处理的。过滤器的处理操作在 next 的最后一环中执行。next 返回的响应 response 是经过过滤器处理的。
如果某个通过 AddBeforeFilterHandler 添加的中间件处理器跳过了 next 而直接返回了结果的话，则返回的 response 也是未经过过滤器处理的。而且如果某个 AddBeforeFilterHandler 添加的中间件处理器跳过了 next，不但过滤器不会执行，而且在它之后使用 AddBeforeFilterHandler 所添加的中间件处理器也不会执行，AddAfterFilterHandler 方法所添加的所有中间件处理器也都不会执行。
而 AddAfterFilterHandler 添加的处理器所收到的 request 都是经过过滤器处理以后的，但它当中使用 next 方法返回的 response 是未经过过滤器处理的。
下面，我们在来看一个结合了压缩过滤器和输入输出缓存中间件的例子。
压缩我们使用 Hprose 过滤器 一章中的 compress_handler.go，这里就不列代码了。
package main
import (
"fmt"
"time"
"sync"
"github.com/hprose/hprose-golang/rpc"
)
type cacheFilter struct {
Cache map[string][]byte
sync.RWMutex
}
func (cf *cacheFilter) handler(
request []byte,
context rpc.Context,
next rpc.NextFilterHandler) (response []byte, err error) {
if context.GetBool("cache") {
var ok bool
cf.RLock()
if response, ok = cf.Cache[string(request)]; ok {
cf.RUnlock()
return response, nil
}
cf.RUnlock()
response, err = next(request, context)
if err != nil {
return
}
cf.Lock()
cf.Cache[string(request)] = response
cf.Unlock()
return response, nil
}
return next(request, context)
}
type sizeFilter struct {
Message string
}
func (sf sizeFilter) handler(
request []byte,
context rpc.Context,
next rpc.NextFilterHandler) (response []byte, err error) {
fmt.Printf("%v request size is %d\r\n", sf.Message, len(request))
response, err = next(request, context)
fmt.Printf("%v response size is %d\r\n", sf.Message, len(response))
return
}
type statFilter struct {
Message string
}
func (sf statFilter) handler(
request []byte,
context rpc.Context,
next rpc.NextFilterHandler) (response []byte, err error) {
start := time.Now()
response, err = next(request, context)
end := time.Now()
fmt.Printf("%v takes %d ms.\r\n", sf.Message, end.Sub(start)/time.Millisecond)
return
}
// TestService is ...
type TestService struct {
Test func([]int) ([]int, error) `userdata:"{\"cache\":true}"`
}
func main() {
server := rpc.NewTCPServer("")
server.AddFunction("test", func(data []int) []int {
return data
}).
AddBeforeFilterHandler(
statFilter{"Server: BeforeFilter"}.handler,
sizeFilter{"Server: Compressed"}.handler,
).
AddFilter(CompressFilter{}).
AddAfterFilterHandler(
statFilter{"Server: AfterFilter"}.handler,
sizeFilter{"Server: Non Compressed"}.handler,
)
server.Handle()
client := rpc.NewClient(server.URI())
client.AddFilter(CompressFilter{}).
AddBeforeFilterHandler(
(&cacheFilter{Cache: make(map[string][]byte)}).handler,
statFilter{"Client: BeforeFilter"}.handler,
sizeFilter{"Client: Compressed"}.handler,
).
AddAfterFilterHandler(
statFilter{"Client: AfterFilter"}.handler,
sizeFilter{"Client: Non Compressed"}.handler,
)
var testService *TestService
client.UseService(&testService)
args := make([]int, 100000)
for i := range args {
args[i] = i
}
result, err := testService.Test(args)
fmt.Println(len(result), err)
result, err = testService.Test(args)
fmt.Println(len(result), err)
client.Close()
server.Close()
}该程序运行结果为：
Client: Compressed request size is 688893
Client: Non Compressed request size is 213244
Server: Compressed request size is 213244
Server: Non Compressed request size is 688893
Server: Non Compressed response size is 688881
Server: AfterFilter takes 5 ms.
Server: Compressed response size is 213223
Server: BeforeFilter takes 38 ms.
Client: Non Compressed response size is 213223
Client: AfterFilter takes 38 ms.
Client: Compressed response size is 688881
Client: BeforeFilter takes 69 ms.
100000 <nil>
100000 <nil>我们可以看到两次的执行结果都出来了，但是中间件的输出内容只有一次。原因就是第二次执行时，缓存中间件将缓存的结果直接返回了。因此后面所有的步骤就都略过了。
通过这个例子，我们可以看出，将 Hprose 中间件和 Hprose 过滤器结合，可以实现非常强大的扩展功能。如果你有什么特殊的需求，直接使用 Hprose 无法实现的话，就考虑一下是否可以添加几个 Hprose 中间件和 Hprose 过滤器吧。