Subversion Repositories openrisc

// 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"

        "io"

        "log"

        "net"

        "net/http"

        "sync"

)

// 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.

type Client struct {

        mutex    sync.Mutex // protects pending, seq, request

        sending  sync.Mutex

        request  Request

        seq      uint64

        codec    ClientCodec

        pending  map[uint64]*Call

        closing  bool

        shutdown bool

}

// 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.

type ClientCodec interface {

        WriteRequest(*Request, interface{}) error

        ReadResponseHeader(*Response) error

        ReadResponseBody(interface{}) error

        Close() error

}

func (client *Client) send(call *Call) {

        client.sending.Lock()

        defer client.sending.Unlock()

        // Register this call.

        client.mutex.Lock()

        if client.shutdown {

                call.Error = ErrShutdown

                client.mutex.Unlock()

                call.done()

                return

        }

        seq := client.seq

        client.seq++

        client.pending[seq] = call

        client.mutex.Unlock()

        // Encode and send the request.

        client.request.Seq = seq

        client.request.ServiceMethod = call.ServiceMethod

        err := client.codec.WriteRequest(&client.request, call.Args)

        if err != nil {

                client.mutex.Lock()

                delete(client.pending, seq)

                client.mutex.Unlock()

                call.Error = err

                call.done()

        }

}

func (client *Client) input() {

        var err error

        var response Response

        for err == nil {

                response = Response{}

                err = client.codec.ReadResponseHeader(&response)

                if err != nil {

                        if err == io.EOF && !client.closing {

                                err = io.ErrUnexpectedEOF

                        }

                        break

                }

                seq := response.Seq

                client.mutex.Lock()

                call := client.pending[seq]

                delete(client.pending, seq)

                client.mutex.Unlock()

                if response.Error == "" {

                        err = client.codec.ReadResponseBody(call.Reply)

                        if err != nil {

                                call.Error = errors.New("reading body " + err.Error())

                        }

                } else {

                        // 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()

        }

        // Terminate pending calls.

        client.sending.Lock()

        client.mutex.Lock()

        client.shutdown = true

        closing := client.closing

        for _, call := range client.pending {

                call.Error = err

                call.done()

        }

        client.mutex.Unlock()

        client.sending.Unlock()

        if 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().

                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.

func NewClient(conn io.ReadWriteCloser) *Client {

        encBuf := bufio.NewWriter(conn)

        client := &gobClientCodec{conn, gob.NewDecoder(conn), gob.NewEncoder(encBuf), encBuf}

        return NewClientWithCodec(client)

}

// NewClientWithCodec is like NewClient but uses the specified

// codec to encode requests and decode responses.

func NewClientWithCodec(codec ClientCodec) *Client {

        client := &Client{

                codec:   codec,

                pending: make(map[uint64]*Call),

        }

        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

        }

        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), 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

        }

        return NewClient(conn), nil

}

func (client *Client) Close() error {

        client.mutex.Lock()

        if client.shutdown || 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) *Call {

        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.Done = done

        client.send(call)

        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 {

        call := <-client.Go(serviceMethod, args, reply, make(chan *Call, 1)).Done

        return call.Error

}