<commit_before>package rpc

import (
	"errors"
	"fmt"
	"io"
	"launchpad.net/juju-core/log"
	"sync"
)

var ErrShutdown = errors.New("connection is shut down")

// 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.
// The params argument to WriteRequest will always be of struct
// type; the result argument to ReadResponseBody will always be
// a non-nil pointer to a struct.
type ClientCodec interface {
	WriteRequest(req *Request, params interface{}) error
	ReadResponseHeader(resp *Response) error
	ReadResponseBody(result interface{}) error
	Close() error
}

// 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 {
	sending sync.Mutex
	codec   ClientCodec
	request Request

	mutex    sync.Mutex // protects the following fields
	reqId    uint64
	pending  map[uint64]*Call
	closing  bool
	shutdown bool
}

// NewClient returns a new Client to handle requests to the set of
// services at the other end of the connection.  The given codec is used
// 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
}

// Call represents an active RPC.
type Call struct {
	Type     string
	Id       string
	Request  string
	Params   interface{}
	Response interface{}
	Error    error
	Done     chan *Call
}

// ServerError represents an error returned from an RPC server.
type ServerError struct {
	Message string
	Code    string
}

func (e *ServerError) Error() string {
	return "server error: " + e.Message
}

func (e *ServerError) ErrorCode() string {
	return e.Code
}

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

func (client *Client) send(call *Call) {
	client.sending.Lock()
	defer client.sending.Unlock()

	// Register this call.
	client.mutex.Lock()
	if client.shutdown || client.closing {
		call.Error = ErrShutdown
		client.mutex.Unlock()
		call.done()
		return
	}

	client.reqId++
	reqId := client.reqId
	client.pending[reqId] = call
	client.mutex.Unlock()

	// Encode and send the request.
	client.request = Request{
		RequestId: reqId,
		Type:      call.Type,
		Id:        call.Id,
		Request:   call.Request,
	}
	params := call.Params
	if params == nil {
		params = struct{}{}
	}
	if err := client.codec.WriteRequest(&client.request, params); err != nil {
		client.mutex.Lock()
		call = client.pending[reqId]<commit_msg>rpc: remove blank line<commit_after>package rpc

import (
	"errors"
	"fmt"
	"io"
	"launchpad.net/juju-core/log"
	"sync"
)

var ErrShutdown = errors.New("connection is shut down")

// 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.
// The params argument to WriteRequest will always be of struct
// type; the result argument to ReadResponseBody will always be
// a non-nil pointer to a struct.
type ClientCodec interface {
	WriteRequest(req *Request, params interface{}) error
	ReadResponseHeader(resp *Response) error
	ReadResponseBody(result interface{}) error
	Close() error
}

// 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 {
	sending sync.Mutex
	codec   ClientCodec
	request Request

	mutex    sync.Mutex // protects the following fields
	reqId    uint64
	pending  map[uint64]*Call
	closing  bool
	shutdown bool
}

// NewClient returns a new Client to handle requests to the set of
// services at the other end of the connection.  The given codec is used
// 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
}

// Call represents an active RPC.
type Call struct {
	Type     string
	Id       string
	Request  string
	Params   interface{}
	Response interface{}
	Error    error
	Done     chan *Call
}

// ServerError represents an error returned from an RPC server.
type ServerError struct {
	Message string
	Code    string
}

func (e *ServerError) Error() string {
	return "server error: " + e.Message
}

func (e *ServerError) ErrorCode() string {
	return e.Code
}

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

func (client *Client) send(call *Call) {
	client.sending.Lock()
	defer client.sending.Unlock()

	// Register this call.
	client.mutex.Lock()
	if client.shutdown || client.closing {
		call.Error = ErrShutdown
		client.mutex.Unlock()
		call.done()
		return
	}
	client.reqId++
	reqId := client.reqId
	client.pending[reqId] = call
	client.mutex.Unlock()

	// Encode and send the request.
	client.request = Request{
		RequestId: reqId,
		Type:      call.Type,
		Id:        call.Id,
		Request:   call.Request,
	}
	params := call.Params
	if params == nil {
		params = struct{}{}
	}
	if err := client.codec.WriteRequest(&client.request, params); err != nil {
		client.mutex.Lock()
		call = client.pending[reqId]
{'fraction_non_alphanumeric': 0.09403285283271874, 'fraction_numerical': 0.002011397921555481, 'mean_word_length': 3.6291698991466252, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1621895', 'n_tokens_mistral': 1843, 'n_tokens_neox': 1789, 'n_words': 658}