vendor/github.com/go-kit/kit/examples/addsvc/pkg/addtransport/http.go

   1 package addtransport
   2
   3 import (
   4         "bytes"
   5         "context"
   6         "encoding/json"
   7         "errors"
   8         "io/ioutil"
   9         "net/http"
  10         "net/url"
  11         "strings"
  12         "time"
  13
  14         "golang.org/x/time/rate"
  15
  16         stdopentracing "github.com/opentracing/opentracing-go"
  17         "github.com/sony/gobreaker"
  18
  19         "github.com/go-kit/kit/circuitbreaker"
  20         "github.com/go-kit/kit/endpoint"
  21         "github.com/go-kit/kit/log"
  22         "github.com/go-kit/kit/ratelimit"
  23         "github.com/go-kit/kit/tracing/opentracing"
  24         httptransport "github.com/go-kit/kit/transport/http"
  25
  26         "github.com/go-kit/kit/examples/addsvc/pkg/addendpoint"
  27         "github.com/go-kit/kit/examples/addsvc/pkg/addservice"
  28 )
  29
  30 // NewHTTPHandler returns an HTTP handler that makes a set of endpoints
  31 // available on predefined paths.
  32 func NewHTTPHandler(endpoints addendpoint.Set, tracer stdopentracing.Tracer, logger log.Logger) http.Handler {
  33         options := []httptransport.ServerOption{
  34                 httptransport.ServerErrorEncoder(errorEncoder),
  35                 httptransport.ServerErrorLogger(logger),
  36         }
  37         m := http.NewServeMux()
  38         m.Handle("/sum", httptransport.NewServer(
  39                 endpoints.SumEndpoint,
  40                 decodeHTTPSumRequest,
  41                 encodeHTTPGenericResponse,
  42                 append(options, httptransport.ServerBefore(opentracing.HTTPToContext(tracer, "Sum", logger)))...,
  43         ))
  44         m.Handle("/concat", httptransport.NewServer(
  45                 endpoints.ConcatEndpoint,
  46                 decodeHTTPConcatRequest,
  47                 encodeHTTPGenericResponse,
  48                 append(options, httptransport.ServerBefore(opentracing.HTTPToContext(tracer, "Concat", logger)))...,
  49         ))
  50         return m
  51 }
  52
  53 // NewHTTPClient returns an AddService backed by an HTTP server living at the
  54 // remote instance. We expect instance to come from a service discovery system,
  55 // so likely of the form "host:port". We bake-in certain middlewares,
  56 // implementing the client library pattern.
  57 func NewHTTPClient(instance string, tracer stdopentracing.Tracer, logger log.Logger) (addservice.Service, error) {
  58         // Quickly sanitize the instance string.
  59         if !strings.HasPrefix(instance, "http") {
  60                 instance = "http://" + instance
  61         }
  62         u, err := url.Parse(instance)
  63         if err != nil {
  64                 return nil, err
  65         }
  66
  67         // We construct a single ratelimiter middleware, to limit the total outgoing
  68         // QPS from this client to all methods on the remote instance. We also
  69         // construct per-endpoint circuitbreaker middlewares to demonstrate how
  70         // that's done, although they could easily be combined into a single breaker
  71         // for the entire remote instance, too.
  72         limiter := ratelimit.NewErroringLimiter(rate.NewLimiter(rate.Every(time.Second), 100))
  73
  74         // Each individual endpoint is an http/transport.Client (which implements
  75         // endpoint.Endpoint) that gets wrapped with various middlewares. If you
  76         // made your own client library, you'd do this work there, so your server
  77         // could rely on a consistent set of client behavior.
  78         var sumEndpoint endpoint.Endpoint
  79         {
  80                 sumEndpoint = httptransport.NewClient(
  81                         "POST",
  82                         copyURL(u, "/sum"),
  83                         encodeHTTPGenericRequest,
  84                         decodeHTTPSumResponse,
  85                         httptransport.ClientBefore(opentracing.ContextToHTTP(tracer, logger)),
  86                 ).Endpoint()
  87                 sumEndpoint = opentracing.TraceClient(tracer, "Sum")(sumEndpoint)
  88                 sumEndpoint = limiter(sumEndpoint)
  89                 sumEndpoint = circuitbreaker.Gobreaker(gobreaker.NewCircuitBreaker(gobreaker.Settings{
  90                         Name:    "Sum",
  91                         Timeout: 30 * time.Second,
  92                 }))(sumEndpoint)
  93         }
  94
  95         // The Concat endpoint is the same thing, with slightly different
  96         // middlewares to demonstrate how to specialize per-endpoint.
  97         var concatEndpoint endpoint.Endpoint
  98         {
  99                 concatEndpoint = httptransport.NewClient(
 100                         "POST",
 101                         copyURL(u, "/concat"),
 102                         encodeHTTPGenericRequest,
 103                         decodeHTTPConcatResponse,
 104                         httptransport.ClientBefore(opentracing.ContextToHTTP(tracer, logger)),
 105                 ).Endpoint()
 106                 concatEndpoint = opentracing.TraceClient(tracer, "Concat")(concatEndpoint)
 107                 concatEndpoint = limiter(concatEndpoint)
 108                 concatEndpoint = circuitbreaker.Gobreaker(gobreaker.NewCircuitBreaker(gobreaker.Settings{
 109                         Name:    "Concat",
 110                         Timeout: 10 * time.Second,
 111                 }))(concatEndpoint)
 112         }
 113
 114         // Returning the endpoint.Set as a service.Service relies on the
 115         // endpoint.Set implementing the Service methods. That's just a simple bit
 116         // of glue code.
 117         return addendpoint.Set{
 118                 SumEndpoint:    sumEndpoint,
 119                 ConcatEndpoint: concatEndpoint,
 120         }, nil
 121 }
 122
 123 func copyURL(base *url.URL, path string) *url.URL {
 124         next := *base
 125         next.Path = path
 126         return &next
 127 }
 128
 129 func errorEncoder(_ context.Context, err error, w http.ResponseWriter) {
 130         w.WriteHeader(err2code(err))
 131         json.NewEncoder(w).Encode(errorWrapper{Error: err.Error()})
 132 }
 133
 134 func err2code(err error) int {
 135         switch err {
 136         case addservice.ErrTwoZeroes, addservice.ErrMaxSizeExceeded, addservice.ErrIntOverflow:
 137                 return http.StatusBadRequest
 138         }
 139         return http.StatusInternalServerError
 140 }
 141
 142 func errorDecoder(r *http.Response) error {
 143         var w errorWrapper
 144         if err := json.NewDecoder(r.Body).Decode(&w); err != nil {
 145                 return err
 146         }
 147         return errors.New(w.Error)
 148 }
 149
 150 type errorWrapper struct {
 151         Error string `json:"error"`
 152 }
 153
 154 // decodeHTTPSumRequest is a transport/http.DecodeRequestFunc that decodes a
 155 // JSON-encoded sum request from the HTTP request body. Primarily useful in a
 156 // server.
 157 func decodeHTTPSumRequest(_ context.Context, r *http.Request) (interface{}, error) {
 158         var req addendpoint.SumRequest
 159         err := json.NewDecoder(r.Body).Decode(&req)
 160         return req, err
 161 }
 162
 163 // decodeHTTPConcatRequest is a transport/http.DecodeRequestFunc that decodes a
 164 // JSON-encoded concat request from the HTTP request body. Primarily useful in a
 165 // server.
 166 func decodeHTTPConcatRequest(_ context.Context, r *http.Request) (interface{}, error) {
 167         var req addendpoint.ConcatRequest
 168         err := json.NewDecoder(r.Body).Decode(&req)
 169         return req, err
 170 }
 171
 172 // decodeHTTPSumResponse is a transport/http.DecodeResponseFunc that decodes a
 173 // JSON-encoded sum response from the HTTP response body. If the response has a
 174 // non-200 status code, we will interpret that as an error and attempt to decode
 175 // the specific error message from the response body. Primarily useful in a
 176 // client.
 177 func decodeHTTPSumResponse(_ context.Context, r *http.Response) (interface{}, error) {
 178         if r.StatusCode != http.StatusOK {
 179                 return nil, errors.New(r.Status)
 180         }
 181         var resp addendpoint.SumResponse
 182         err := json.NewDecoder(r.Body).Decode(&resp)
 183         return resp, err
 184 }
 185
 186 // decodeHTTPConcatResponse is a transport/http.DecodeResponseFunc that decodes
 187 // a JSON-encoded concat response from the HTTP response body. If the response
 188 // has a non-200 status code, we will interpret that as an error and attempt to
 189 // decode the specific error message from the response body. Primarily useful in
 190 // a client.
 191 func decodeHTTPConcatResponse(_ context.Context, r *http.Response) (interface{}, error) {
 192         if r.StatusCode != http.StatusOK {
 193                 return nil, errors.New(r.Status)
 194         }
 195         var resp addendpoint.ConcatResponse
 196         err := json.NewDecoder(r.Body).Decode(&resp)
 197         return resp, err
 198 }
 199
 200 // encodeHTTPGenericRequest is a transport/http.EncodeRequestFunc that
 201 // JSON-encodes any request to the request body. Primarily useful in a client.
 202 func encodeHTTPGenericRequest(_ context.Context, r *http.Request, request interface{}) error {
 203         var buf bytes.Buffer
 204         if err := json.NewEncoder(&buf).Encode(request); err != nil {
 205                 return err
 206         }
 207         r.Body = ioutil.NopCloser(&buf)
 208         return nil
 209 }
 210
 211 // encodeHTTPGenericResponse is a transport/http.EncodeResponseFunc that encodes
 212 // the response as JSON to the response writer. Primarily useful in a server.
 213 func encodeHTTPGenericResponse(ctx context.Context, w http.ResponseWriter, response interface{}) error {
 214         if f, ok := response.(addendpoint.Failer); ok && f.Failed() != nil {
 215                 errorEncoder(ctx, f.Failed(), w)
 216                 return nil
 217         }
 218         w.Header().Set("Content-Type", "application/json; charset=utf-8")
 219         return json.NewEncoder(w).Encode(response)
 220 }