package server + example

addsvc/.gitignore

@@ -0,0 +1 @@
+addsvc

addsvc/add.go

@@ -0,0 +1,22 @@
+package main
+
+// Add is the abstract definition of what this service does.
+// It could easily be an interface type with multiple methods.
+type Add func(int64, int64) int64
+
+func pureAdd(a, b int64) int64 { return a + b }
+
+func addVia(r Resource) Add {
+	return func(a, b int64) int64 {
+		return r.Value(a) + r.Value(b)
+	}
+}
+
+// Resource represents some dependency, outside of our control.
+type Resource interface {
+	Value(int64) int64
+}
+
+type mockResource struct{}
+
+func (mockResource) Value(i int64) int64 { return i }

addsvc/endpoint.go

@@ -0,0 +1,32 @@
+package main
+
+import (
+	"github.com/peterbourgon/gokit/server"
+	"golang.org/x/net/context"
+)
+
+// makeEndpoint returns a server.Endpoint wrapping the passed Add. If Add were
+// an interface with multiple methods, we'd need individual endpoints for
+// each.
+//
+// This function is just boiler-plate; in theory, it could be generated.
+func makeEndpoint(a Add) server.Endpoint {
+	return func(ctx context.Context, req server.Request) (server.Response, error) {
+		select {
+		case <-ctx.Done():
+			return nil, server.ErrContextCanceled
+		default:
+		}
+
+		addReq, ok := req.(*request)
+		if !ok {
+			return nil, server.ErrBadCast
+		}
+
+		v := a(addReq.A, addReq.B)
+
+		return response{
+			V: v,
+		}, nil
+	}
+}

addsvc/enhancements.go

@@ -0,0 +1,22 @@
+package main
+
+import (
+	"encoding/json"
+	"io"
+	"time"
+)
+
+func logging(w io.Writer, add Add) Add {
+	return func(a, b int64) (v int64) {
+		defer func(begin time.Time) {
+			json.NewEncoder(w).Encode(map[string]interface{}{
+				"a":      a,
+				"b":      b,
+				"result": v,
+				"took":   time.Since(begin),
+			})
+		}(time.Now())
+		v = add(a, b)
+		return
+	}
+}

addsvc/grpc_binding.go

@@ -0,0 +1,50 @@
+package main
+
+import (
+	"time"
+
+	"github.com/peterbourgon/gokit/addsvc/pb"
+	"github.com/peterbourgon/gokit/metrics"
+	"github.com/peterbourgon/gokit/server"
+	"golang.org/x/net/context"
+)
+
+// A binding wraps an Endpoint so that it's usable by a transport. grpcBinding
+// makes an Endpoint usable over gRPC.
+type grpcBinding struct{ server.Endpoint }
+
+// Add implements the proto3 AddServer by forwarding to the wrapped Endpoint.
+func (b grpcBinding) Add(ctx context.Context, req *pb.AddRequest) (*pb.AddReply, error) {
+	addReq := request{req.A, req.B}
+	r, err := b.Endpoint(ctx, addReq)
+	if err != nil {
+		return nil, err
+	}
+
+	resp, ok := r.(*response)
+	if !ok {
+		return nil, server.ErrBadCast
+	}
+
+	return &pb.AddReply{
+		V: resp.V,
+	}, nil
+}
+
+func grpcInstrument(requests metrics.Counter, duration metrics.Histogram) func(pb.AddServer) pb.AddServer {
+	return func(next pb.AddServer) pb.AddServer {
+		return grpcInstrumented{requests, duration, next}
+	}
+}
+
+type grpcInstrumented struct {
+	requests metrics.Counter
+	duration metrics.Histogram
+	next     pb.AddServer
+}
+
+func (i grpcInstrumented) Add(ctx context.Context, req *pb.AddRequest) (*pb.AddReply, error) {
+	i.requests.Add(1)
+	defer func(begin time.Time) { i.duration.Observe(time.Since(begin).Nanoseconds()) }(time.Now())
+	return i.next.Add(ctx, req)
+}

addsvc/httpjson_binding.go

@@ -0,0 +1,80 @@
+package main
+
+import (
+	"encoding/json"
+	"io"
+	"net/http"
+	"time"
+
+	"golang.org/x/net/context"
+
+	"github.com/peterbourgon/gokit/metrics"
+	"github.com/peterbourgon/gokit/server"
+	"github.com/peterbourgon/gokit/server/zipkin"
+	"github.com/peterbourgon/gokit/transport/codec"
+)
+
+// jsonCodec decodes and encodes requests and responses respectively as JSON.
+// It requires that the (package main) request and response structs support
+// JSON de/serialization.
+//
+// This type is mostly boiler-plate; in theory, it could be generated.
+type jsonCodec struct{}
+
+func (jsonCodec) Decode(ctx context.Context, r io.Reader) (server.Request, context.Context, error) {
+	var req request
+	err := json.NewDecoder(r).Decode(&req)
+	return &req, ctx, err
+}
+
+func (jsonCodec) Encode(w io.Writer, resp server.Response) error {
+	return json.NewEncoder(w).Encode(resp)
+}
+
+// A binding wraps an Endpoint so that it's usable by a transport. httpBinding
+// makes an Endpoint usable over HTTP. It combines a parent context, a codec,
+// and an endpoint to expose. It implements http.Handler by decoding a request
+// from the HTTP request body, and encoding a response to the response writer.
+type httpBinding struct {
+	context.Context        // parent context
+	codec.Codec            // how to decode requests and encode responses
+	contentType     string // what we report as the response ContentType
+	server.Endpoint        // the endpoint being bound
+}
+
+func (b httpBinding) ServeHTTP(w http.ResponseWriter, r *http.Request) {
+	// Perform HTTP-specific context amendments.
+	b.Context = zipkin.GetHeaders(b.Context, r.Header)
+
+	// Decode request.
+	req, ctx, err := b.Codec.Decode(b.Context, r.Body)
+	if err != nil {
+		http.Error(w, err.Error(), http.StatusBadRequest)
+		return
+	}
+	b.Context = ctx
+
+	// Execute RPC.
+	resp, err := b.Endpoint(b.Context, req)
+	if err != nil {
+		http.Error(w, err.Error(), http.StatusInternalServerError)
+		return
+	}
+
+	// Encode response.
+	w.Header().Set("Content-Type", b.contentType)
+	if err := b.Codec.Encode(w, resp); err != nil {
+		http.Error(w, err.Error(), http.StatusInternalServerError)
+		return
+	}
+}
+
+func httpInstrument(requests metrics.Counter, duration metrics.Histogram) func(http.Handler) http.Handler {
+	return func(next http.Handler) http.Handler {
+		return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+			requests.Add(1)
+			defer func(begin time.Time) { duration.Observe(time.Since(begin).Nanoseconds()) }(time.Now())
+			next.ServeHTTP(w, r)
+		})
+	}
+}

addsvc/main.go

@@ -0,0 +1,115 @@
+package main
+
+import (
+	"flag"
+	"fmt"
+	"io/ioutil"
+	"log"
+	"net"
+	"net/http"
+	"os"
+	"os/signal"
+	"syscall"
+	"time"
+
+	"golang.org/x/net/context"
+	"google.golang.org/grpc"
+
+	"github.com/peterbourgon/gokit/addsvc/pb"
+	"github.com/peterbourgon/gokit/metrics"
+	"github.com/peterbourgon/gokit/metrics/expvar"
+	"github.com/peterbourgon/gokit/metrics/statsd"
+	"github.com/peterbourgon/gokit/server"
+	"github.com/peterbourgon/gokit/server/zipkin"
+	"github.com/peterbourgon/gokit/transport/http/cors"
+)
+
+func main() {
+	var (
+		httpJSONAddr = flag.String("http.json.addr", ":8001", "Address for HTTP/JSON server")
+		grpcTCPAddr  = flag.String("grpc.tcp.addr", ":8002", "Address for gRPC (TCP) server")
+	)
+	flag.Parse()
+
+	// Our business and operational domain
+	var a Add
+	a = pureAdd
+	a = logging(logWriter{}, a)
+
+	// `package server` domain
+	var e server.Endpoint
+	e = makeEndpoint(a)
+	e = server.Gate(zipkin.RequireInContext)(e) // must have Zipkin headers
+	// e = server.ChainableEnhancement(arg1, arg2, e)
+
+	// `package metrics` domain
+	requests := metrics.NewMultiCounter(
+		expvar.NewCounter("requests"),
+		statsd.NewCounter(ioutil.Discard, "requests", time.Second),
+	)
+	duration := metrics.NewMultiHistogram(
+		expvar.NewHistogram("duration_ns", 0, 100000000, 3),
+		statsd.NewHistogram(ioutil.Discard, "duration_ns", time.Second),
+	)
+
+	// Mechanical stuff
+	root := context.Background()
+	errc := make(chan error)
+
+	go func() {
+		errc <- interrupt()
+	}()
+
+	// Transport: gRPC
+	go func() {
+		ln, err := net.Listen("tcp", *grpcTCPAddr)
+		if err != nil {
+			errc <- err
+			return
+		}
+		s := grpc.NewServer() // uses its own context?
+		field := metrics.Field{Key: "transport", Value: "grpc"}
+
+		var addServer pb.AddServer
+		addServer = grpcBinding{e}
+		addServer = grpcInstrument(requests.With(field), duration.With(field))(addServer)
+		// Note that this will always fail, because the Endpoint is gated on
+		// Zipkin headers, and we don't extract them from the gRPC request.
+
+		pb.RegisterAddServer(s, addServer)
+		log.Printf("gRPC server on TCP %s", *grpcTCPAddr)
+		errc <- s.Serve(ln)
+	}()
+
+	// Transport: HTTP/JSON
+	go func() {
+		ctx, cancel := context.WithCancel(root)
+		defer cancel()
+		mux := http.NewServeMux()
+		field := metrics.Field{Key: "transport", Value: "http"}
+
+		var handler http.Handler
+		handler = httpBinding{ctx, jsonCodec{}, "application/json", e}
+		handler = httpInstrument(requests.With(field), duration.With(field))(handler)
+		handler = cors.Middleware(cors.MaxAge(5 * time.Minute))(handler)
+
+		mux.Handle("/add", handler)
+		log.Printf("HTTP/JSON server on %s", *httpJSONAddr)
+		errc <- http.ListenAndServe(*httpJSONAddr, mux)
+	}()
+
+	log.Fatal(<-errc)
+}
+
+type logWriter struct{}
+
+func (logWriter) Write(p []byte) (int, error) {
+	log.Printf("%s", p)
+	return len(p), nil
+}
+
+func interrupt() error {
+	c := make(chan os.Signal)
+	signal.Notify(c, syscall.SIGINT, syscall.SIGTERM)
+	return fmt.Errorf("%s", <-c)
+}