mirror of
https://github.com/duanhf2012/origin.git
synced 2026-02-21 12:14:45 +08:00
提交origin2.0版本
This commit is contained in:
duanhf2012
514
rpc/client.go
514
rpc/client.go
@@ -1,360 +1,208 @@
|
||||
// Copyright 2009 The Go Authors. All rights reserved.
|
||||
// Use of this source code is governed by a BSD-style
|
||||
// license that can be found in the LICENSE file.
|
||||
|
||||
package rpc
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"encoding/gob"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"log"
|
||||
"net"
|
||||
"net/http"
|
||||
"github.com/duanhf2012/originnet/log"
|
||||
"github.com/duanhf2012/originnet/network"
|
||||
"math"
|
||||
"reflect"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
// ServerError represents an error that has been returned from
|
||||
// the remote side of the RPC connection.
|
||||
type ServerError string
|
||||
|
||||
func (e ServerError) Error() string {
|
||||
return string(e)
|
||||
}
|
||||
|
||||
var ErrShutdown = errors.New("connection is shut down")
|
||||
|
||||
// Call represents an active RPC.
|
||||
type Call struct {
|
||||
ServiceMethod string // The name of the service and method to call.
|
||||
Args interface{} // The argument to the function (*struct).
|
||||
Reply interface{} // The reply from the function (*struct).
|
||||
Error error // After completion, the error status.
|
||||
Done chan *Call // Strobes when call is complete.
|
||||
}
|
||||
|
||||
// Client represents an RPC Client.
|
||||
// There may be multiple outstanding Calls associated
|
||||
// with a single Client, and a Client may be used by
|
||||
// multiple goroutines simultaneously.
|
||||
type Client struct {
|
||||
codec ClientCodec
|
||||
blocalhost bool
|
||||
network.TCPClient
|
||||
conn *network.TCPConn
|
||||
|
||||
reqMutex sync.Mutex // protects following
|
||||
request Request
|
||||
|
||||
mutex sync.Mutex // protects following
|
||||
seq uint64
|
||||
pending map[uint64]*Call
|
||||
closing bool // user has called Close
|
||||
shutdown bool // server has told us to stop
|
||||
bPipe bool
|
||||
pendingLock sync.RWMutex
|
||||
startSeq uint64
|
||||
pending map[uint64]*Call
|
||||
}
|
||||
|
||||
// A ClientCodec implements writing of RPC requests and
|
||||
// reading of RPC responses for the client side of an RPC session.
|
||||
// The client calls WriteRequest to write a request to the connection
|
||||
// and calls ReadResponseHeader and ReadResponseBody in pairs
|
||||
// to read responses. The client calls Close when finished with the
|
||||
// connection. ReadResponseBody may be called with a nil
|
||||
// argument to force the body of the response to be read and then
|
||||
// discarded.
|
||||
// See NewClient's comment for information about concurrent access.
|
||||
type ClientCodec interface {
|
||||
WriteRequest(*Request, interface{}) error
|
||||
ReadResponseHeader(*Response) error
|
||||
ReadResponseBody(interface{}) error
|
||||
|
||||
Close() error
|
||||
func (slf *Client) NewClientAgent(conn *network.TCPConn) network.Agent {
|
||||
slf.conn = conn
|
||||
return slf
|
||||
}
|
||||
|
||||
func (client *Client) IsClosed() bool {
|
||||
client.reqMutex.Lock()
|
||||
defer client.reqMutex.Unlock()
|
||||
return client.shutdown || client.closing
|
||||
}
|
||||
|
||||
func (client *Client) send(call *Call, queueMode bool) {
|
||||
client.reqMutex.Lock()
|
||||
defer client.reqMutex.Unlock()
|
||||
|
||||
// Register this call.
|
||||
client.mutex.Lock()
|
||||
if client.shutdown || client.closing {
|
||||
client.mutex.Unlock()
|
||||
call.Error = ErrShutdown
|
||||
call.done()
|
||||
return
|
||||
func (slf *Client) Connect(addr string) error {
|
||||
slf.Addr = addr
|
||||
if strings.Index(addr,"localhost") == 0 {
|
||||
slf.blocalhost = true
|
||||
return nil
|
||||
}
|
||||
seq := client.seq
|
||||
client.seq++
|
||||
client.pending[seq] = call
|
||||
client.mutex.Unlock()
|
||||
slf.ConnNum = 1
|
||||
slf.ConnectInterval = time.Second*2
|
||||
slf.PendingWriteNum = 10000
|
||||
slf.AutoReconnect = true
|
||||
slf.LenMsgLen =2
|
||||
slf.MinMsgLen = 2
|
||||
slf.MaxMsgLen = math.MaxUint16
|
||||
slf.NewAgent =slf.NewClientAgent
|
||||
slf.LittleEndian = LittleEndian
|
||||
|
||||
// Encode and send the request.
|
||||
client.request.Seq = seq
|
||||
client.request.ServiceMethod = call.ServiceMethod
|
||||
client.request.QueueMode = queueMode
|
||||
err := client.codec.WriteRequest(&client.request, call.Args)
|
||||
if err != nil {
|
||||
client.mutex.Lock()
|
||||
call = client.pending[seq]
|
||||
delete(client.pending, seq)
|
||||
client.mutex.Unlock()
|
||||
if call != nil {
|
||||
call.Error = err
|
||||
call.done()
|
||||
}
|
||||
slf.pendingLock.Lock()
|
||||
for _,v := range slf.pending {
|
||||
v.Err = fmt.Errorf("node is disconnect.")
|
||||
v.done <- v
|
||||
}
|
||||
slf.pending = map[uint64]*Call{}
|
||||
slf.pendingLock.Unlock()
|
||||
slf.Start()
|
||||
return nil
|
||||
}
|
||||
|
||||
func (client *Client) input() {
|
||||
var err error
|
||||
var response Response
|
||||
for err == nil {
|
||||
response = Response{}
|
||||
err = client.codec.ReadResponseHeader(&response)
|
||||
if err != nil {
|
||||
break
|
||||
}
|
||||
seq := response.Seq
|
||||
client.mutex.Lock()
|
||||
call := client.pending[seq]
|
||||
delete(client.pending, seq)
|
||||
client.mutex.Unlock()
|
||||
|
||||
switch {
|
||||
case call == nil:
|
||||
// We've got no pending call. That usually means that
|
||||
// WriteRequest partially failed, and call was already
|
||||
// removed; response is a server telling us about an
|
||||
// error reading request body. We should still attempt
|
||||
// to read error body, but there's no one to give it to.
|
||||
err = client.codec.ReadResponseBody(nil)
|
||||
if err != nil {
|
||||
err = errors.New("reading error body: " + err.Error())
|
||||
}
|
||||
case response.Error != "":
|
||||
// We've got an error response. Give this to the request;
|
||||
// any subsequent requests will get the ReadResponseBody
|
||||
// error if there is one.
|
||||
call.Error = ServerError(response.Error)
|
||||
err = client.codec.ReadResponseBody(nil)
|
||||
if err != nil {
|
||||
err = errors.New("reading error body: " + err.Error())
|
||||
}
|
||||
call.done()
|
||||
default:
|
||||
err = client.codec.ReadResponseBody(call.Reply)
|
||||
if err != nil {
|
||||
call.Error = errors.New("reading body " + err.Error())
|
||||
}
|
||||
if client.bPipe {
|
||||
err = nil
|
||||
}
|
||||
call.done()
|
||||
}
|
||||
}
|
||||
// Terminate pending calls.
|
||||
client.reqMutex.Lock()
|
||||
client.mutex.Lock()
|
||||
client.shutdown = true
|
||||
closing := client.closing
|
||||
if err == io.EOF {
|
||||
if closing {
|
||||
err = ErrShutdown
|
||||
} else {
|
||||
err = io.ErrUnexpectedEOF
|
||||
}
|
||||
}
|
||||
for _, call := range client.pending {
|
||||
call.Error = err
|
||||
call.done()
|
||||
}
|
||||
client.mutex.Unlock()
|
||||
client.reqMutex.Unlock()
|
||||
if debugLog && err != io.EOF && !closing {
|
||||
log.Println("rpc: client protocol error:", err)
|
||||
}
|
||||
}
|
||||
|
||||
func (call *Call) done() {
|
||||
select {
|
||||
case call.Done <- call:
|
||||
// ok
|
||||
default:
|
||||
// We don't want to block here. It is the caller's responsibility to make
|
||||
// sure the channel has enough buffer space. See comment in Go().
|
||||
if debugLog {
|
||||
log.Println("rpc: discarding Call reply due to insufficient Done chan capacity")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// NewClient returns a new Client to handle requests to the
|
||||
// set of services at the other end of the connection.
|
||||
// It adds a buffer to the write side of the connection so
|
||||
// the header and payload are sent as a unit.
|
||||
//
|
||||
// The read and write halves of the connection are serialized independently,
|
||||
// so no interlocking is required. However each half may be accessed
|
||||
// concurrently so the implementation of conn should protect against
|
||||
// concurrent reads or concurrent writes.
|
||||
func NewClient(conn io.ReadWriteCloser, isPipe bool) *Client {
|
||||
encBuf := bufio.NewWriter(conn)
|
||||
client := &gobClientCodec{conn, gob.NewDecoder(conn), gob.NewEncoder(encBuf), encBuf}
|
||||
return NewClientWithCodec(client, isPipe)
|
||||
}
|
||||
|
||||
// NewClientWithCodec is like NewClient but uses the specified
|
||||
// codec to encode requests and decode responses.
|
||||
func NewClientWithCodec(codec ClientCodec, isPipe bool) *Client {
|
||||
client := &Client{
|
||||
codec: codec,
|
||||
pending: make(map[uint64]*Call),
|
||||
bPipe: isPipe,
|
||||
}
|
||||
go client.input()
|
||||
return client
|
||||
}
|
||||
|
||||
type gobClientCodec struct {
|
||||
rwc io.ReadWriteCloser
|
||||
dec *gob.Decoder
|
||||
enc *gob.Encoder
|
||||
encBuf *bufio.Writer
|
||||
}
|
||||
|
||||
func (c *gobClientCodec) WriteRequest(r *Request, body interface{}) (err error) {
|
||||
if err = c.enc.Encode(r); err != nil {
|
||||
return
|
||||
}
|
||||
if err = c.enc.Encode(body); err != nil {
|
||||
return
|
||||
}
|
||||
return c.encBuf.Flush()
|
||||
}
|
||||
|
||||
func (c *gobClientCodec) ReadResponseHeader(r *Response) error {
|
||||
return c.dec.Decode(r)
|
||||
}
|
||||
|
||||
func (c *gobClientCodec) ReadResponseBody(body interface{}) error {
|
||||
return c.dec.Decode(body)
|
||||
}
|
||||
|
||||
func (c *gobClientCodec) Close() error {
|
||||
return c.rwc.Close()
|
||||
}
|
||||
|
||||
// DialHTTP connects to an HTTP RPC server at the specified network address
|
||||
// listening on the default HTTP RPC path.
|
||||
func DialHTTP(network, address string) (*Client, error) {
|
||||
return DialHTTPPath(network, address, DefaultRPCPath)
|
||||
}
|
||||
|
||||
// DialHTTPPath connects to an HTTP RPC server
|
||||
// at the specified network address and path.
|
||||
func DialHTTPPath(network, address, path string) (*Client, error) {
|
||||
var err error
|
||||
conn, err := net.Dial(network, address)
|
||||
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
tcpconn, _ := conn.(*net.TCPConn)
|
||||
tcpconn.SetNoDelay(true)
|
||||
|
||||
io.WriteString(conn, "CONNECT "+path+" HTTP/1.0\n\n")
|
||||
|
||||
// Require successful HTTP response
|
||||
// before switching to RPC protocol.
|
||||
resp, err := http.ReadResponse(bufio.NewReader(conn), &http.Request{Method: "CONNECT"})
|
||||
if err == nil && resp.Status == connected {
|
||||
return NewClient(conn, false), nil
|
||||
}
|
||||
if err == nil {
|
||||
err = errors.New("unexpected HTTP response: " + resp.Status)
|
||||
}
|
||||
conn.Close()
|
||||
return nil, &net.OpError{
|
||||
Op: "dial-http",
|
||||
Net: network + " " + address,
|
||||
Addr: nil,
|
||||
Err: err,
|
||||
}
|
||||
}
|
||||
|
||||
// Dial connects to an RPC server at the specified network address.
|
||||
func Dial(network, address string) (*Client, error) {
|
||||
conn, err := net.Dial(network, address)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
tcpconn, _ := conn.(*net.TCPConn)
|
||||
tcpconn.SetNoDelay(true)
|
||||
return NewClient(conn, false), nil
|
||||
}
|
||||
|
||||
func DialTimeOut(network, address string, timeout time.Duration) (*Client, error) {
|
||||
conn, err := net.DialTimeout(network, address, timeout)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
tcpconn, _ := conn.(*net.TCPConn)
|
||||
tcpconn.SetNoDelay(true)
|
||||
return NewClient(conn, false), nil
|
||||
}
|
||||
|
||||
// Close calls the underlying codec's Close method. If the connection is already
|
||||
// shutting down, ErrShutdown is returned.
|
||||
func (client *Client) Close() error {
|
||||
client.mutex.Lock()
|
||||
if client.closing {
|
||||
client.mutex.Unlock()
|
||||
return ErrShutdown
|
||||
}
|
||||
client.closing = true
|
||||
client.mutex.Unlock()
|
||||
return client.codec.Close()
|
||||
}
|
||||
|
||||
// Go invokes the function asynchronously. It returns the Call structure representing
|
||||
// the invocation. The done channel will signal when the call is complete by returning
|
||||
// the same Call object. If done is nil, Go will allocate a new channel.
|
||||
// If non-nil, done must be buffered or Go will deliberately crash.
|
||||
func (client *Client) Go(serviceMethod string, args interface{}, reply interface{}, done chan *Call, queueMode bool) *Call {
|
||||
func (slf *Client) AsycGo(rpcHandler IRpcHandler,mutiCoroutine bool,serviceMethod string,callback reflect.Value, args interface{},replyParam interface{}) error {
|
||||
call := new(Call)
|
||||
call.ServiceMethod = serviceMethod
|
||||
call.Args = args
|
||||
call.Reply = reply
|
||||
if done == nil {
|
||||
done = make(chan *Call, 10) // buffered.
|
||||
} else {
|
||||
// If caller passes done != nil, it must arrange that
|
||||
// done has enough buffer for the number of simultaneous
|
||||
// RPCs that will be using that channel. If the channel
|
||||
// is totally unbuffered, it's best not to run at all.
|
||||
if cap(done) == 0 {
|
||||
log.Panic("rpc: done channel is unbuffered")
|
||||
}
|
||||
call.Reply = replyParam
|
||||
call.callback = &callback
|
||||
call.rpcHandler = rpcHandler
|
||||
|
||||
request := &RpcRequest{}
|
||||
request.NoReply = false
|
||||
request.MutiCoroutine = mutiCoroutine
|
||||
call.Arg = args
|
||||
slf.pendingLock.Lock()
|
||||
slf.startSeq += 1
|
||||
call.Seq = slf.startSeq
|
||||
request.Seq = slf.startSeq
|
||||
slf.pending[call.Seq] = call
|
||||
slf.pendingLock.Unlock()
|
||||
|
||||
request.ServiceMethod = serviceMethod
|
||||
var herr error
|
||||
request.InParam,herr = processor.Marshal(args)
|
||||
if herr != nil {
|
||||
return herr
|
||||
}
|
||||
call.Done = done
|
||||
client.send(call, queueMode)
|
||||
|
||||
bytes,err := processor.Marshal(request)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if slf.conn == nil {
|
||||
return fmt.Errorf("Rpc server is disconnect,call %s is fail!",serviceMethod)
|
||||
}
|
||||
err = slf.conn.WriteMsg(bytes)
|
||||
if err != nil {
|
||||
call.Err = err
|
||||
}
|
||||
|
||||
return call.Err
|
||||
}
|
||||
|
||||
func (slf *Client) Go(noReply bool,mutiCoroutine bool,serviceMethod string, args interface{},reply interface{}) *Call {
|
||||
call := new(Call)
|
||||
call.done = make(chan *Call,1)
|
||||
call.Reply = reply
|
||||
|
||||
request := &RpcRequest{}
|
||||
request.MutiCoroutine = mutiCoroutine
|
||||
request.NoReply = noReply
|
||||
call.Arg = args
|
||||
slf.pendingLock.Lock()
|
||||
slf.startSeq += 1
|
||||
call.Seq = slf.startSeq
|
||||
request.Seq = slf.startSeq
|
||||
slf.pending[call.Seq] = call
|
||||
slf.pendingLock.Unlock()
|
||||
|
||||
request.ServiceMethod = serviceMethod
|
||||
var herr error
|
||||
request.InParam,herr = processor.Marshal(args)
|
||||
if herr != nil {
|
||||
call.Err = herr
|
||||
return call
|
||||
}
|
||||
|
||||
bytes,err := processor.Marshal(request)
|
||||
if err != nil {
|
||||
call.Err = err
|
||||
return call
|
||||
}
|
||||
|
||||
err = slf.conn.WriteMsg(bytes)
|
||||
if err != nil {
|
||||
call.Err = err
|
||||
}
|
||||
|
||||
return call
|
||||
}
|
||||
|
||||
// Call invokes the named function, waits for it to complete, and returns its error status.
|
||||
func (client *Client) Call(serviceMethod string, args interface{}, reply interface{}) error {
|
||||
select {
|
||||
case call := <-client.Go(serviceMethod, args, reply, make(chan *Call, 1), false).Done:
|
||||
return call.Error
|
||||
case <-time.After(30 * time.Second):
|
||||
type RequestHandler func(Returns interface{},Err error)
|
||||
|
||||
type RpcRequest struct {
|
||||
//packhead
|
||||
Seq uint64 // sequence number chosen by client
|
||||
ServiceMethod string // format: "Service.Method"
|
||||
NoReply bool //是否需要返回
|
||||
MutiCoroutine bool // 是否多协程模式
|
||||
|
||||
//packbody
|
||||
InParam []byte
|
||||
localReply interface{}
|
||||
localParam interface{} //本地调用的参数列表
|
||||
requestHandle RequestHandler
|
||||
|
||||
callback *reflect.Value
|
||||
}
|
||||
|
||||
type RpcResponse struct {
|
||||
//head
|
||||
Seq uint64 // sequence number chosen by client
|
||||
Err error
|
||||
|
||||
//returns
|
||||
Returns []byte
|
||||
}
|
||||
|
||||
|
||||
func (slf *Client) Run(){
|
||||
for {
|
||||
bytes,err := slf.conn.ReadMsg()
|
||||
if err != nil {
|
||||
slf.Close()
|
||||
slf.Start()
|
||||
}
|
||||
//1.解析head
|
||||
respone := &RpcResponse{}
|
||||
err = processor.Unmarshal(bytes,respone)
|
||||
if err != nil {
|
||||
log.Error("rpcClient Unmarshal head error,error:%+v",err)
|
||||
continue
|
||||
}
|
||||
|
||||
slf.pendingLock.Lock()
|
||||
v,ok := slf.pending[respone.Seq]
|
||||
if ok == false {
|
||||
log.Error("rpcClient cannot find seq %d in pending",respone.Seq)
|
||||
slf.pendingLock.Unlock()
|
||||
}else {
|
||||
delete(slf.pending,respone.Seq)
|
||||
slf.pendingLock.Unlock()
|
||||
|
||||
err = processor.Unmarshal(respone.Returns,v.Reply)
|
||||
if err != nil {
|
||||
log.Error("rpcClient Unmarshal body error,error:%+v",err)
|
||||
continue
|
||||
}
|
||||
|
||||
if v.callback.IsValid() {
|
||||
v.rpcHandler.(*RpcHandler).callResponeCallBack<-v
|
||||
}else{
|
||||
//发送至接受者
|
||||
v.done <- v
|
||||
}
|
||||
|
||||
}
|
||||
}
|
||||
|
||||
//call := <-client.Go(serviceMethod, args, reply, make(chan *Call, 1)).Done
|
||||
return errors.New(fmt.Sprintf("Call RPC %s is time out 30s", serviceMethod))
|
||||
}
|
||||
|
||||
func (slf *Client) OnClose(){
|
||||
if slf.blocalhost== false{
|
||||
//关闭时，重新连接
|
||||
slf.Start()
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user