srpc/client.go

package srpc

import (
	"bytes"
	"errors"
	"fmt"
	"io"
	"net"
	"sync"
	"time"
)

type Client struct {
	sync.RWMutex
	ed      IEncoderDecoder
	conn    IRPCConn
	running bool
}

func (client *Client) Call(funcName string, args ...interface{}) (ret []byte, err error) {
	var b bytes.Buffer

	enc := client.ed.NewEncoder(&b)
	for _, a := range args {
		enc.Encode(a)
	}

	payload := b.Bytes()
	req := RPCRequest{funcName, payload}

	client.RLock()
	defer client.RUnlock()

	if err = client.conn.Send(&req); err != nil {
		return nil, err
	}

	var call *RPCCall
	if call, err = client.conn.Receive(); err != nil {
		return nil, err
	}

	var response *RPCResponse
	var ok bool
	if response, ok = call.Payload.(*RPCResponse); !ok {
		return nil, errors.New("srpc - Expected response")
	}

	if response.Status != RPCOK {
		err = errors.New("srpc - Response contained error: '" + response.Error + "'")
	}

	return response.Payload, err
}

func (client *Client) Close() {
	client.Lock()
	defer client.Unlock()

	client.conn.Send(new(RPCClose))
	client.running = false
}

func (client *Client) NewDecoder(r io.Reader) IDecoder {
	return client.ed.NewDecoder(r)
}

func NewClient(conn net.Conn) *Client {
	ret := &Client{sync.RWMutex{}, NewEncoderDecoder(), NewNetConn(conn, NewEncoderDecoder()), true}
	go func() {
		for {
			fmt.Println("heartbeat")
			ret.Lock()

			if ret.running == false {
				ret.Unlock()
				return
			}
			ret.conn.Send(new(RPCHeartbeat))
			if _, err := ret.conn.Receive(); err != nil {
				ret.conn.Send(new(RPCClose))
				ret.Unlock()
				return
			}
			fmt.Println("got heartbeat")
			ret.Unlock()
			time.Sleep(15 * time.Second)
		}
	}()

	return ret
}

type DialFunc func(addr string) (conn io.ReadWriteCloser, err error)

func unixDial(addr string) (conn io.ReadWriteCloser, err error) {
	if conn, err = net.Dial("unix", addr); err != nil {
		return nil, err
	}

	return conn, nil
}