client.go raw

   1  package xmlrpc
   2  
   3  import (
   4  	"errors"
   5  	"fmt"
   6  	"io"
   7  	"io/ioutil"
   8  	"net/http"
   9  	"net/http/cookiejar"
  10  	"net/rpc"
  11  	"net/url"
  12  	"strconv"
  13  	"sync"
  14  	"time"
  15  )
  16  
  17  type Client struct {
  18  	*rpc.Client
  19  }
  20  
  21  // clientCodec is rpc.ClientCodec interface implementation.
  22  type clientCodec struct {
  23  	// url presents url of xmlrpc service
  24  	url *url.URL
  25  
  26  	// httpClient works with HTTP protocol
  27  	httpClient *http.Client
  28  
  29  	// cookies stores cookies received on last request
  30  	cookies http.CookieJar
  31  
  32  	// responses presents map of active requests. It is required to return request id, that
  33  	// rpc.Client can mark them as done.
  34  
  35  	responsesMu sync.RWMutex
  36  	responses   map[uint64]*http.Response
  37  	response Response
  38  
  39  	// ready presents channel, that is used to link request and it`s response.
  40  	ready chan uint64
  41  	// close notifies codec is closed.
  42  	close chan uint64
  43  
  44  }
  45  
  46  func (codec *clientCodec) WriteRequest(request *rpc.Request, args interface{}) (err error) {
  47  	httpRequest, err := NewRequest(codec.url.String(), request.ServiceMethod, args)
  48  	if err != nil {
  49  		return err
  50  	}
  51  
  52  	if codec.cookies != nil {
  53  		for _, cookie := range codec.cookies.Cookies(codec.url) {
  54  			httpRequest.AddCookie(cookie)
  55  		}
  56  	}
  57  
  58  
  59  	httpResponse, err := codec.httpClient.Do(httpRequest)
  60  
  61  	if err != nil {
  62  		return err
  63  	}
  64  
  65  	if codec.cookies != nil {
  66  		codec.cookies.SetCookies(codec.url, httpResponse.Cookies())
  67  	}
  68  
  69  	codec.responsesMu.Lock()
  70  	codec.responses[request.Seq] = httpResponse
  71  	codec.responsesMu.Unlock()
  72  	codec.ready <- request.Seq
  73  
  74  	return nil
  75  }
  76  
  77  func (codec *clientCodec) ReadResponseHeader(response *rpc.Response) (err error) {
  78  	var seq uint64
  79  	select {
  80  		case seq = <-codec.ready:
  81  		case <-codec.close:
  82  			return errors.New("codec is closed")
  83  	}
  84  	response.Seq = seq
  85  
  86  	codec.responsesMu.RLock()
  87  	httpResponse := codec.responses[seq]
  88  	delete(codec.responses, seq)
  89  	codec.responsesMu.RUnlock()
  90  
  91  	defer httpResponse.Body.Close()
  92  
  93  	contentLength := httpResponse.ContentLength
  94  	if contentLength == -1 {
  95  		if ntcoentLengthHeader, ok := httpResponse.Header["Ntcoent-Length"]; ok {
  96  			ntcoentLength, err := strconv.ParseInt(ntcoentLengthHeader[0], 10, 64)
  97  			if err == nil {
  98  				contentLength = ntcoentLength
  99  			}
 100  		}
 101  	}
 102  
 103  	var respData []byte
 104  	if contentLength != -1 {
 105  		respData = make([]byte, contentLength)
 106  		_, err = io.ReadFull(httpResponse.Body, respData)
 107  	} else {
 108  		respData, err = ioutil.ReadAll(httpResponse.Body)
 109  	}
 110  
 111  	if err != nil {
 112  		response.Error = err.Error()
 113  		return nil
 114  	}
 115  
 116  
 117  	resp := NewResponse(respData, httpResponse.StatusCode)
 118  
 119  	if resp.Failed() {
 120  		err := resp.Err()
 121  		response.Error = fmt.Sprintf("%v", err)
 122  		return err
 123  
 124  	}
 125  	codec.response = *resp
 126  
 127  	if httpResponse.StatusCode < 200 || httpResponse.StatusCode >= 300 {
 128  		return &XmlRpcError{HttpStatusCode: httpResponse.StatusCode}
 129  	}
 130  
 131  	return nil
 132  }
 133  
 134  func (codec *clientCodec) ReadResponseBody(v interface{}) (err error) {
 135  	if v == nil {
 136  		return nil
 137  	}
 138  	return codec.response.Unmarshal(v)
 139  }
 140  
 141  func (codec *clientCodec) Close() error {
 142  	if transport, ok := codec.httpClient.Transport.(*http.Transport); ok {
 143  		transport.CloseIdleConnections()
 144  	}
 145  
 146  	close(codec.close)
 147  	return nil
 148  }
 149  
 150  // NewClient returns instance of rpc.Client object, that is used to send request to xmlrpc service.
 151  func NewClient(requrl string, transport http.RoundTripper, timeout time.Duration) (*Client, error) {
 152  	if transport == nil {
 153  		transport = http.DefaultTransport
 154  	}
 155  
 156  	httpClient := &http.Client{
 157  		Transport: transport,
 158  		Timeout:   timeout,
 159  	}
 160  
 161  	jar, err := cookiejar.New(nil)
 162  
 163  	if err != nil {
 164  		return nil, err
 165  	}
 166  
 167  	u, err := url.Parse(requrl)
 168  
 169  	if err != nil {
 170  		return nil, err
 171  	}
 172  
 173  	codec := clientCodec{
 174  		url:        u,
 175  		httpClient: httpClient,
 176  		ready:      make(chan uint64),
 177  		close:      make(chan uint64),
 178  		responses:  make(map[uint64]*http.Response),
 179  		cookies:    jar,
 180  	}
 181  
 182  	return &Client{rpc.NewClientWithCodec(&codec)}, nil
 183  }
 184