<commit_before>package typhon

import (
	"io"
	"io/ioutil"
	"net/http"
	"sync"
	"time"

	"github.com/facebookgo/httpcontrol"
	"github.com/mondough/terrors"
	"golang.org/x/net/context"
)

var (
	// Client is used to send all requests by default. It can be overridden globally but MUST only be done before use
	// takes place; access is not synchronised.
	Client              Service = BareClient
	httpClientTransport         = &httpcontrol.Transport{
		Proxy:                 http.ProxyFromEnvironment,
		DisableKeepAlives:     false,
		DisableCompression:    false,
		MaxIdleConnsPerHost:   10,
		DialTimeout:           10 * time.Second,
		DialKeepAlive:         10 * time.Minute,
		ResponseHeaderTimeout: time.Minute,
		RequestTimeout:        time.Hour,
		RetryAfterTimeout:     false,
		MaxTries:              6}
	httpClient = &http.Client{ // shared for all outbound requests
		Timeout:   time.Hour,
		Transport: httpClientTransport}
)

type ResponseFuture struct {
	r      Response
	cancel context.CancelFunc
	mtx    sync.RWMutex
	done   <-chan struct{}
}

func (f *ResponseFuture) WaitC() <-chan struct{} {
	return f.done
}

func (f *ResponseFuture) Response() Response {
	<-f.WaitC()
	f.mtx.RLock()
	defer f.mtx.RUnlock()
	return f.r
}

func (f *ResponseFuture) Cancel() {
	f.cancel()
}

func BareClient(req Request) Response {
	// If the request is cancelled, cancel the request
	// Go 1.7's http library has native context support, so this can go away
	httpReq := &req.Request
	done := make(chan struct{})
	cancel := make(chan struct{})
	superCancel := (<-chan struct{})(cancel)
	if httpReq.Cancel != nil { // If there already was a cancel channel, wrap it
		superCancel = httpReq.Cancel
	}
	httpReq.Cancel = cancel
	go func() {
		select {
		case <-done:
		case <-httpReq.Cancel:
		case <-superCancel:
			close(cancel)
		case <-req.Context.Done():
			close(cancel)
		}
	}()

	httpRsp, err := httpClient.Do(httpReq)
	close(done)

	// Read the response in its entirety and close the Response body here; this protects us from callers that forget to
	// call Close() but does not allow streaming responses.
	// @TODO: Streaming client?
	if httpRsp != nil && httpRsp.Body != nil {
		buf := &bufCloser{}
		io.Copy(buf, httpRsp.Body)
		httpRsp.Body.Close()
		httpRsp.Body = ioutil.NopCloser(buf)
	}

	return Response{
		Response: httpRsp,
		Error:    terrors.Wrap(err, nil)}
}

func SendVia(req Request, svc Service) *ResponseFuture {
	ctx, cancel := context.WithCancel(req.Context)
	req.Context = ctx
	done := make(chan struct{})
	f := &ResponseFuture{
		cancel: cancel,
		done:   done}
	go func() {
		defer cancel() // if already cancelled on escape, this is a no-op
		defer close(done)
		rsp := svc(req)
		f.mtx.RLock()
		defer f.mtx.RUnlock()
		f.r = rsp
	}()
	return f
}

func Send(req Request) *ResponseFuture {
	return SendVia(req, Client)
}
<commit_msg>Refactor request cancellation<commit_after>package typhon

import (
	"bytes"
	"io/ioutil"
	"net/http"
	"sync"
	"time"

	"github.com/facebookgo/httpcontrol"
	log "github.com/mondough/slog"
	"github.com/mondough/terrors"
	"golang.org/x/net/context"
)

var (
	// Client is used to send all requests by default. It can be overridden globally but MUST only be done before use
	// takes place; access is not synchronised.
	Client              = BareClient
	httpClientTransport = &httpcontrol.Transport{
		Proxy:                 http.ProxyFromEnvironment,
		DisableKeepAlives:     false,
		DisableCompression:    false,
		MaxIdleConnsPerHost:   10,
		DialTimeout:           10 * time.Second,
		DialKeepAlive:         10 * time.Minute,
		ResponseHeaderTimeout: time.Minute,
		RequestTimeout:        time.Hour,
		RetryAfterTimeout:     false,
		MaxTries:              6}
	httpClient = &http.Client{ // shared for all outbound requests
		Timeout:   time.Hour,
		Transport: httpClientTransport}
)

type ResponseFuture struct {
	r      Response
	cancel context.CancelFunc
	mtx    sync.RWMutex
	done   <-chan struct{}
}

func (f *ResponseFuture) WaitC() <-chan struct{} {
	return f.done
}

func (f *ResponseFuture) Response() Response {
	<-f.WaitC()
	f.mtx.RLock()
	defer f.mtx.RUnlock()
	return f.r
}

func (f *ResponseFuture) Cancel() {
	f.cancel()
}

// httpCancellationFilter ties together the context cancellation and the cancel channel of net/http.Request. It is
// incorporated into BareClient by default.
// @TODO: Go 1.7's http library has native context support, so this can go away
func httpCancellationFilter(req Request, svc Service) Response {
	ctx, ctxCancel := context.WithCancel(req.Context)
	defer ctxCancel()

	// When the context is cancelled, propagate this to net/http
	// If the caller set the net/http Cancel channel, allow this to be used too
	httpCancel := make(chan struct{})
	httpSuperCancel := req.Request.Cancel
	req.Request.Cancel = httpCancel
	go func() {
		select {
		case <-ctx.Done():
			close(httpCancel)
		case <-httpSuperCancel:
			close(httpCancel)
		case <-httpCancel:
		}
	}()

	req.Context = ctx
	return svc(req)
}

func BareClient(req Request) Response {
	return httpCancellationFilter(req, func(req Request) Response {
		httpRsp, err := httpClient.Do(&req.Request)

		// Read the response in its entirety and close the Response body here; this protects us from callers that forget to
		// call Close() but does not allow streaming responses.
		// @TODO: Streaming client?
		if httpRsp != nil && httpRsp.Body != nil {
			buf, err := ioutil.ReadAll(httpRsp.Body)
			httpRsp.Body.Close()
			if err != nil {
				log.Warn(req, "Error reading response body: %v", err)
			}
			httpRsp.Body = ioutil.NopCloser(bytes.NewReader(buf))
		}

		return Response{
			Response: httpRsp,
			Error:    terrors.Wrap(err, nil)}
	})
}

func SendVia(req Request, svc Service) *ResponseFuture {
	ctx, cancel := context.WithCancel(req.Context)
	req.Context = ctx
	f := &ResponseFuture{
		done:   ctx.Done(),
		cancel: cancel}
	go func() {
		defer cancel() // if already cancelled on escape, this is a no-op
		rsp := svc(req)
		f.mtx.RLock()
		f.r = rsp
		f.mtx.RUnlock()
	}()
	return f
}

func Send(req Request) *ResponseFuture {
	return SendVia(req, Client)
}

{'fraction_non_alphanumeric': 0.11404796867351934, 'fraction_numerical': 0.002936857562408223, 'mean_word_length': 3.3413597733711047, 'pattern_counts': {'":': 0, '<': 17, '<?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': '7499531', 'n_tokens_mistral': 2182, 'n_tokens_neox': 2051, 'n_words': 604}