NeonXP/jsonrpc2
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity

Middlewares and options

Browse source
This commit is contained in:
Alexander Kiryukhin 2022-05-22 16:37:48 +03:00
parent c74596c6a6
commit 4a81eff217
No known key found for this signature in database
GPG key ID: 6DF7A2910D0699E9
10 changed files with 253 additions and 91 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

30
README.md
View file

@ -13,19 +13,29 @@ Go 1.18+ required
## Usage (http transport)
1. Create JSON-RPC server:
1. Create JSON-RPC server with options:
```go
import "go.neonxp.dev/jsonrpc2/rpc"
...
s := rpc.New()
s := rpc.New(
rpc.WithTransport(&transport.HTTP{
Bind: ":8000", // Port to bind
CORSOrigin: "*", // CORS origin
TLS: &tls.Config{}, // Optional TLS config (default nil)
Parallel: true, // Allow parallel run batch methods (default false)
}),
//Other options like transports/middlewares...
)
```
2. Add required transport(s):
```go
import "go.neonxp.dev/jsonrpc2/transport"
...
s.AddTransport(&transport.HTTP{Bind: ":8000"})
s.AddTransport(&transport.TCP{Bind: ":3000"})
s.Use(
rpc.WithTransport(&transport.TCP{Bind: ":3000"}),
//...
)
```
3. Write handler:
@ -72,10 +82,16 @@ import (
)
func main() {
s := rpc.New()
s := rpc.New(
rpc.WithLogger(rpc.StdLogger), // Optional logger
rpc.WithTransport(&transport.HTTP{Bind: ":8000"}), // HTTP transport
)
s.AddTransport(&transport.HTTP{Bind: ":8000"}) // HTTP transport
s.AddTransport(&transport.TCP{Bind: ":3000"}) // TCP transport
// Set options after constructor
s.Use(
rpc.WithTransport(&transport.TCP{Bind: ":3000"}), // TCP transport
rpc.WithMiddleware(rpc.LoggerMiddleware(rpc.StdLogger)), // Logger middleware
)
s.Register("multiply", rpc.H(Multiply))
s.Register("divide", rpc.H(Divide))

13
example/main.go
View file

@ -12,10 +12,15 @@ import (
)
func main() {
s := rpc.New()
s.AddTransport(&transport.HTTP{Bind: ":8000"})
s.AddTransport(&transport.TCP{Bind: ":3000"})
s := rpc.New(
rpc.WithLogger(rpc.StdLogger),
rpc.WithTransport(&transport.HTTP{Bind: ":8000", CORSOrigin: "*"}),
)
// Set options after constructor
s.Use(
rpc.WithTransport(&transport.TCP{Bind: ":3000"}),
rpc.WithMiddleware(rpc.LoggerMiddleware(rpc.StdLogger)),
)
s.Register("multiply", rpc.H(Multiply))
s.Register("divide", rpc.H(Divide))

36
example/test.http
View file

@ -11,18 +11,6 @@ Content-Type: application/json
}
date: Sat, 21 May 2022 16:53:31 GMT
content-length: 36
content-type: text/plain; charset=utf-8
connection: close
HTTP/1.1 200 - OK
date: Sat, 21 May 2022 16:53:31 GMT
content-length: 36
content-type: text/plain; charset=utf-8
connection: close
###
POST http://localhost:8000/
Content-Type: application/json
@ -37,18 +25,6 @@ Content-Type: application/json
}
date: Sat, 21 May 2022 16:53:51 GMT
content-length: 52
content-type: text/plain; charset=utf-8
connection: close
HTTP/1.1 200 - OK
date: Sat, 21 May 2022 16:53:51 GMT
content-length: 52
content-type: text/plain; charset=utf-8
connection: close
### Batch request
POST http://localhost:8000/
Content-Type: application/json
@ -68,15 +44,3 @@ Content-Type: application/json
{"foo": "boo"}
{"jsonrpc": "2.0", "method": "foo.get", "params": {"name": "myself"}, "id": "5"}
{"jsonrpc": "2.0", "method": "get_data", "id": "9"}
date: Sat, 21 May 2022 17:18:17 GMT
content-type: text/plain; charset=utf-8
connection: close
transfer-encoding: chunked
HTTP/1.1 200 - OK
date: Sat, 21 May 2022 17:18:17 GMT
content-type: text/plain; charset=utf-8
connection: close
transfer-encoding: chunked

3
rpc/meta.go
View file

@ -1,3 +0,0 @@
package rpc
type Meta map[string]interface{}

43
rpc/middleware.go Normal file
View file

@ -0,0 +1,43 @@
//Package rpc provides abstract rpc server
//
//Copyright (C) 2022 Alexander Kiryukhin <i@neonxp.dev>
//
//This file is part of go.neonxp.dev/jsonrpc2 project.
//
//This program is free software: you can redistribute it and/or modify
//it under the terms of the GNU General Public License as published by
//the Free Software Foundation, either version 3 of the License, or
//(at your option) any later version.
//
//This program is distributed in the hope that it will be useful,
//but WITHOUT ANY WARRANTY; without even the implied warranty of
//MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
//GNU General Public License for more details.
//
//You should have received a copy of the GNU General Public License
//along with this program. If not, see <https://www.gnu.org/licenses/>.
package rpc
import (
"context"
"strings"
"time"
)
type Middleware func(handler RpcHandler) RpcHandler
type RpcHandler func(ctx context.Context, req *RpcRequest) *RpcResponse
func LoggerMiddleware(logger Logger) Middleware {
return func(handler RpcHandler) RpcHandler {
return func(ctx context.Context, req *RpcRequest) *RpcResponse {
t1 := time.Now().UnixMicro()
resp := handler(ctx, req)
t2 := time.Now().UnixMicro()
args := strings.ReplaceAll(string(req.Params), "\n", "")
logger.Logf("rpc call=%s, args=%s, take=%dμs", req.Method, args, (t2 - t1))
return resp
}
}
}

42
rpc/options.go Normal file
View file

@ -0,0 +1,42 @@
//Package rpc provides abstract rpc server
//
//Copyright (C) 2022 Alexander Kiryukhin <i@neonxp.dev>
//
//This file is part of go.neonxp.dev/jsonrpc2 project.
//
//This program is free software: you can redistribute it and/or modify
//it under the terms of the GNU General Public License as published by
//the Free Software Foundation, either version 3 of the License, or
//(at your option) any later version.
//
//This program is distributed in the hope that it will be useful,
//but WITHOUT ANY WARRANTY; without even the implied warranty of
//MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
//GNU General Public License for more details.
//
//You should have received a copy of the GNU General Public License
//along with this program. If not, see <https://www.gnu.org/licenses/>.
package rpc
import "go.neonxp.dev/jsonrpc2/transport"
type Option func(s *RpcServer)
func WithTransport(transport transport.Transport) Option {
return func(s *RpcServer) {
s.transports = append(s.transports, transport)
}
}
func WithMiddleware(mw Middleware) Option {
return func(s *RpcServer) {
s.middlewares = append(s.middlewares, mw)
}
}
func WithLogger(l Logger) Option {
return func(s *RpcServer) {
s.logger = l
}
}

86
rpc/server.go
View file

@ -33,20 +33,29 @@ import (
const version = "2.0"
type RpcServer struct {
Logger Logger
IgnoreNotifications bool
handlers map[string]Handler
transports []transport.Transport
mu sync.RWMutex
logger Logger
handlers map[string]Handler
middlewares []Middleware
transports []transport.Transport
mu sync.RWMutex
}
func New() *RpcServer {
return &RpcServer{
Logger: nopLogger{},
IgnoreNotifications: true,
handlers: map[string]Handler{},
transports: []transport.Transport{},
mu: sync.RWMutex{},
func New(opts ...Option) *RpcServer {
s := &RpcServer{
logger: nopLogger{},
handlers: map[string]Handler{},
transports: []transport.Transport{},
mu: sync.RWMutex{},
}
for _, opt := range opts {
opt(s)
}
return s
}
func (r *RpcServer) Use(opts ...Option) {
for _, opt := range opts {
opt(r)
}
}
@ -56,10 +65,6 @@ func (r *RpcServer) Register(method string, handler Handler) {
r.handlers[method] = handler
}
func (r *RpcServer) AddTransport(transport transport.Transport) {
r.transports = append(r.transports, transport)
}
func (r *RpcServer) Run(ctx context.Context) error {
eg, ctx := errgroup.WithContext(ctx)
for _, t := range r.transports {
@ -70,25 +75,27 @@ func (r *RpcServer) Run(ctx context.Context) error {
return eg.Wait()
}
func (r *RpcServer) Resolve(ctx context.Context, rd io.Reader, w io.Writer) {
func (r *RpcServer) Resolve(ctx context.Context, rd io.Reader, w io.Writer, parallel bool) {
dec := json.NewDecoder(rd)
enc := json.NewEncoder(w)
mu := sync.Mutex{}
wg := sync.WaitGroup{}
for {
req := new(rpcRequest)
req := new(RpcRequest)
if err := dec.Decode(req); err != nil {
if err == io.EOF {
break
}
r.Logger.Logf("Can't read body: %v", err)
r.logger.Logf("Can't read body: %v", err)
WriteError(ErrCodeParseError, enc)
break
}
wg.Add(1)
go func(req *rpcRequest) {
defer wg.Done()
resp := r.callMethod(ctx, req)
exec := func() {
h := r.callMethod
for _, m := range r.middlewares {
h = m(h)
}
resp := h(ctx, req)
if req.Id == nil {
// notification request
return
@ -96,23 +103,34 @@ func (r *RpcServer) Resolve(ctx context.Context, rd io.Reader, w io.Writer) {
mu.Lock()
defer mu.Unlock()
if err := enc.Encode(resp); err != nil {
r.Logger.Logf("Can't write response: %v", err)
r.logger.Logf("Can't write response: %v", err)
WriteError(ErrCodeInternalError, enc)
}
if w, canFlush := w.(Flusher); canFlush {
w.Flush()
}
}(req)
}
if parallel {
wg.Add(1)
go func(req *RpcRequest) {
defer wg.Done()
exec()
}(req)
} else {
exec()
}
}
if parallel {
wg.Wait()
}
wg.Wait()
}
func (r *RpcServer) callMethod(ctx context.Context, req *rpcRequest) *rpcResponse {
func (r *RpcServer) callMethod(ctx context.Context, req *RpcRequest) *RpcResponse {
r.mu.RLock()
h, ok := r.handlers[req.Method]
r.mu.RUnlock()
if !ok {
return &rpcResponse{
return &RpcResponse{
Jsonrpc: version,
Error: ErrorFromCode(ErrCodeMethodNotFound),
Id: req.Id,
@ -120,14 +138,14 @@ func (r *RpcServer) callMethod(ctx context.Context, req *rpcRequest) *rpcRespons
}
resp, err := h(ctx, req.Params)
if err != nil {
r.Logger.Logf("User error %v", err)
return &rpcResponse{
r.logger.Logf("User error %v", err)
return &RpcResponse{
Jsonrpc: version,
Error: err,
Id: req.Id,
}
}
return &rpcResponse{
return &RpcResponse{
Jsonrpc: version,
Result: resp,
Id: req.Id,
@ -135,20 +153,20 @@ func (r *RpcServer) callMethod(ctx context.Context, req *rpcRequest) *rpcRespons
}
func WriteError(code int, enc *json.Encoder) {
enc.Encode(rpcResponse{
enc.Encode(RpcResponse{
Jsonrpc: version,
Error: ErrorFromCode(code),
})
}
type rpcRequest struct {
type RpcRequest struct {
Jsonrpc string `json:"jsonrpc"`
Method string `json:"method"`
Params json.RawMessage `json:"params"`
Id any `json:"id"`
}
type rpcResponse struct {
type RpcResponse struct {
Jsonrpc string `json:"jsonrpc"`
Result json.RawMessage `json:"result,omitempty"`
Error error `json:"error,omitempty"`

37
transport/http.go
View file

@ -1,3 +1,22 @@
//Package transport provides transports for rpc server
//
//Copyright (C) 2022 Alexander Kiryukhin <i@neonxp.dev>
//
//This file is part of go.neonxp.dev/jsonrpc2 project.
//
//This program is free software: you can redistribute it and/or modify
//it under the terms of the GNU General Public License as published by
//the Free Software Foundation, either version 3 of the License, or
//(at your option) any later version.
//
//This program is distributed in the hope that it will be useful,
//but WITHOUT ANY WARRANTY; without even the implied warranty of
//MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
//GNU General Public License for more details.
//
//You should have received a copy of the GNU General Public License
//along with this program. If not, see <https://www.gnu.org/licenses/>.
package transport
import (
@ -8,21 +27,33 @@ import (
)
type HTTP struct {
Bind string
TLS *tls.Config
Bind string
TLS *tls.Config
CORSOrigin string
Parallel bool
}
func (h *HTTP) Run(ctx context.Context, resolver Resolver) error {
srv := http.Server{
Addr: h.Bind,
Handler: http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
if r.Method == http.MethodOptions && h.CORSOrigin != "" {
w.Header().Set("Access-Control-Allow-Origin", h.CORSOrigin)
w.Header().Set("Access-Control-Allow-Headers", "Content-Type")
w.WriteHeader(http.StatusOK)
return
}
if r.Method != http.MethodPost {
w.WriteHeader(http.StatusMethodNotAllowed)
return
}
if h.CORSOrigin != "" {
w.Header().Set("Access-Control-Allow-Origin", h.CORSOrigin)
w.Header().Set("Access-Control-Allow-Headers", "Content-Type")
}
w.Header().Add("Content-Type", "application/json")
w.WriteHeader(http.StatusOK)
resolver.Resolve(ctx, r.Body, w)
resolver.Resolve(ctx, r.Body, w, h.Parallel)
}),
BaseContext: func(l net.Listener) context.Context {
return ctx

33
transport/tcp.go
View file

@ -1,3 +1,22 @@
//Package transport provides transports for rpc server
//
//Copyright (C) 2022 Alexander Kiryukhin <i@neonxp.dev>
//
//This file is part of go.neonxp.dev/jsonrpc2 project.
//
//This program is free software: you can redistribute it and/or modify
//it under the terms of the GNU General Public License as published by
//the Free Software Foundation, either version 3 of the License, or
//(at your option) any later version.
//
//This program is distributed in the hope that it will be useful,
//but WITHOUT ANY WARRANTY; without even the implied warranty of
//MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
//GNU General Public License for more details.
//
//You should have received a copy of the GNU General Public License
//along with this program. If not, see <https://www.gnu.org/licenses/>.
package transport
import (
@ -6,17 +25,25 @@ import (
)
type TCP struct {
Bind string
Bind string
Parallel bool
}
func (t *TCP) Run(ctx context.Context, resolver Resolver) error {
ln, _ := net.Listen("tcp", t.Bind)
ln, err := net.Listen("tcp", t.Bind)
if err != nil {
return err
}
go func() {
<-ctx.Done()
ln.Close()
}()
for {
conn, err := ln.Accept()
if err != nil {
return err
}
go resolver.Resolve(ctx, conn, conn)
go resolver.Resolve(ctx, conn, conn, t.Parallel)
}
}

21
transport/transport.go
View file

@ -1,3 +1,22 @@
//Package transport provides transports for rpc server
//
//Copyright (C) 2022 Alexander Kiryukhin <i@neonxp.dev>
//
//This file is part of go.neonxp.dev/jsonrpc2 project.
//
//This program is free software: you can redistribute it and/or modify
//it under the terms of the GNU General Public License as published by
//the Free Software Foundation, either version 3 of the License, or
//(at your option) any later version.
//
//This program is distributed in the hope that it will be useful,
//but WITHOUT ANY WARRANTY; without even the implied warranty of
//MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
//GNU General Public License for more details.
//
//You should have received a copy of the GNU General Public License
//along with this program. If not, see <https://www.gnu.org/licenses/>.
package transport
import (
@ -10,5 +29,5 @@ type Transport interface {
}
type Resolver interface {
Resolve(context.Context, io.Reader, io.Writer)
Resolve(ctx context.Context, reader io.Reader, writer io.Writer, isParallel bool)
}