First

server/buffer.go

@@ -0,0 +1,14 @@
+package server
+
+import (
+	"io"
+)
+
+type buffer struct {
+	io.Reader
+	io.Writer
+}
+
+func (b *buffer) Close() error {
+	return nil
+}

server/context.go

@@ -0,0 +1,35 @@
+package server
+
+import (
+	"time"
+
+	"code.google.com/p/go.net/context"
+)
+
+type ctx struct{}
+
+func (ctx *ctx) Deadline() (deadline time.Time, ok bool) {
+	return time.Time{}, false
+}
+
+func (ctx *ctx) Done() <-chan struct{} {
+	return nil
+}
+
+func (ctx *ctx) Err() error {
+	return nil
+}
+
+func (ctx *ctx) Value(key interface{}) interface{} {
+	return nil
+}
+
+func newContext(parent context.Context, s *serverContext) context.Context {
+	return context.WithValue(parent, "serverContext", s)
+}
+
+// return server.Context
+func NewContext(ctx context.Context) (Context, bool) {
+	c, ok := ctx.Value("serverContext").(*serverContext)
+	return c, ok
+}

server/headers.go

@@ -0,0 +1,8 @@
+package server
+
+type Headers interface {
+	Add(string, string)
+	Del(string)
+	Get(string) string
+	Set(string, string)
+}

server/health_checker.go

@@ -0,0 +1,21 @@
+package server
+
+import (
+	"io"
+	"net/http"
+	"net/url"
+)
+
+func registerHealthChecker(mux *http.ServeMux) {
+	req := &http.Request{
+		Method: "GET",
+		URL: &url.URL{
+			Path: HealthPath,
+		},
+	}
+	if _, path := mux.Handler(req); path != HealthPath {
+		mux.HandleFunc(HealthPath, func(w http.ResponseWriter, r *http.Request) {
+			io.WriteString(w, "ok")
+		})
+	}
+}

server/receiver.go

@@ -0,0 +1,6 @@
+package server
+
+type Receiver interface {
+	Name() string
+	Handler() interface{}
+}

server/request.go

@@ -0,0 +1,6 @@
+package server
+
+type Request interface {
+	Headers() Headers
+	Session(string) string
+}

server/rpc_receiver.go

@@ -0,0 +1,29 @@
+package server
+
+type RpcReceiver struct {
+	name    string
+	handler interface{}
+}
+
+func newRpcReceiver(name string, handler interface{}) *RpcReceiver {
+	return &RpcReceiver{
+		name:    name,
+		handler: handler,
+	}
+}
+
+func (r *RpcReceiver) Name() string {
+	return r.name
+}
+
+func (r *RpcReceiver) Handler() interface{} {
+	return r.handler
+}
+
+func NewRpcReceiver(handler interface{}) *RpcReceiver {
+	return newRpcReceiver("", handler)
+}
+
+func NewNamedRpcReceiver(name string, handler interface{}) *RpcReceiver {
+	return newRpcReceiver(name, handler)
+}

server/rpc_server.go

@@ -0,0 +1,215 @@
+package server
+
+import (
+	"bytes"
+	"fmt"
+	"io/ioutil"
+	"net"
+	"net/http"
+	"runtime/debug"
+	"strconv"
+	"sync"
+
+	"github.com/asim/go-micro/errors"
+	log "github.com/cihub/seelog"
+	rpc "github.com/youtube/vitess/go/rpcplus"
+	js "github.com/youtube/vitess/go/rpcplus/jsonrpc"
+	pb "github.com/youtube/vitess/go/rpcplus/pbrpc"
+)
+
+type RpcServer struct {
+	mtx     sync.RWMutex
+	rpc     *rpc.Server
+	address string
+	exit    chan chan error
+}
+
+var (
+	HealthPath = "/_status/health"
+	RpcPath    = "/_rpc"
+)
+
+func executeRequestSafely(c *serverContext, r *http.Request) {
+	defer func() {
+		if x := recover(); x != nil {
+			log.Criticalf("Panicked on request: %v", r)
+			log.Criticalf("%v: %v", x, string(debug.Stack()))
+			err := errors.InternalServerError("go.micro.server", "Unexpected error")
+			c.WriteHeader(500)
+			c.Write([]byte(err.Error()))
+		}
+	}()
+
+	http.DefaultServeMux.ServeHTTP(c, r)
+}
+
+func (s *RpcServer) handler(w http.ResponseWriter, r *http.Request) {
+	c := &serverContext{
+		req:       &serverRequest{r},
+		outHeader: w.Header(),
+	}
+
+	ctxs.Lock()
+	ctxs.m[r] = c
+	ctxs.Unlock()
+	defer func() {
+		ctxs.Lock()
+		delete(ctxs.m, r)
+		ctxs.Unlock()
+	}()
+
+	// Patch up RemoteAddr so it looks reasonable.
+	if addr := r.Header.Get("X-Forwarded-For"); len(addr) > 0 {
+		r.RemoteAddr = addr
+	} else {
+		// Should not normally reach here, but pick a sensible default anyway.
+		r.RemoteAddr = "127.0.0.1"
+	}
+	// The address in the headers will most likely be of these forms:
+	//	123.123.123.123
+	//	2001:db8::1
+	// net/http.Request.RemoteAddr is specified to be in "IP:port" form.
+	if _, _, err := net.SplitHostPort(r.RemoteAddr); err != nil {
+		// Assume the remote address is only a host; add a default port.
+		r.RemoteAddr = net.JoinHostPort(r.RemoteAddr, "80")
+	}
+
+	executeRequestSafely(c, r)
+	c.outHeader = nil // make sure header changes aren't respected any more
+
+	// Avoid nil Write call if c.Write is never called.
+	if c.outCode != 0 {
+		w.WriteHeader(c.outCode)
+	}
+	if c.outBody != nil {
+		w.Write(c.outBody)
+	}
+}
+
+func (s *RpcServer) Address() string {
+	s.mtx.RLock()
+	defer s.mtx.RUnlock()
+	return s.address
+}
+
+func (s *RpcServer) ServeHTTP(w http.ResponseWriter, req *http.Request) {
+	serveCtx := getServerContext(req)
+
+	// TODO: get user scope from context
+	// check access
+
+	if req.Method != "POST" {
+		err := errors.BadRequest("go.micro.server", "Method not allowed")
+		http.Error(w, err.Error(), http.StatusMethodNotAllowed)
+		return
+	}
+	defer req.Body.Close()
+
+	b, err := ioutil.ReadAll(req.Body)
+	if err != nil {
+		errr := errors.InternalServerError("go.micro.server", fmt.Sprintf("Error reading request body: %v", err))
+		w.WriteHeader(500)
+		w.Write([]byte(errr.Error()))
+		log.Errorf("Erroring reading request body: %v", err)
+		return
+	}
+
+	rbq := bytes.NewBuffer(b)
+	rsp := bytes.NewBuffer(nil)
+	defer rsp.Reset()
+	defer rbq.Reset()
+
+	buf := &buffer{
+		rbq,
+		rsp,
+	}
+
+	var cc rpc.ServerCodec
+	switch req.Header.Get("Content-Type") {
+	case "application/octet-stream":
+		cc = pb.NewServerCodec(buf)
+	case "application/json":
+		cc = js.NewServerCodec(buf)
+	default:
+		err = errors.InternalServerError("go.micro.server", fmt.Sprintf("Unsupported content-type: %v", req.Header.Get("Content-Type")))
+		w.WriteHeader(500)
+		w.Write([]byte(err.Error()))
+		return
+	}
+
+	ctx := newContext(&ctx{}, serveCtx)
+	err = s.rpc.ServeRequestWithContext(ctx, cc)
+	if err != nil {
+		// This should not be possible.
+		w.WriteHeader(500)
+		w.Write([]byte(err.Error()))
+		log.Errorf("Erroring serving request: %v", err)
+		return
+	}
+
+	w.Header().Set("Content-Type", req.Header.Get("Content-Type"))
+	w.Header().Set("Content-Length", strconv.Itoa(rsp.Len()))
+	w.Write(rsp.Bytes())
+}
+
+func (s *RpcServer) Init() error {
+	log.Debugf("Rpc handler %s", RpcPath)
+	http.Handle(RpcPath, s)
+	return nil
+}
+
+func (s *RpcServer) NewReceiver(handler interface{}) Receiver {
+	return newRpcReceiver("", handler)
+}
+
+func (s *RpcServer) NewNamedReceiver(name string, handler interface{}) Receiver {
+	return newRpcReceiver(name, handler)
+}
+
+func (s *RpcServer) Register(r Receiver) error {
+	if len(r.Name()) > 0 {
+		s.rpc.RegisterName(r.Name(), r.Handler())
+		return nil
+	}
+
+	s.rpc.Register(r.Handler())
+	return nil
+}
+
+func (s *RpcServer) Start() error {
+	registerHealthChecker(http.DefaultServeMux)
+
+	l, err := net.Listen("tcp", s.address)
+	if err != nil {
+		return err
+	}
+
+	log.Debugf("Listening on %s", l.Addr().String())
+
+	s.mtx.Lock()
+	s.address = l.Addr().String()
+	s.mtx.Unlock()
+
+	go http.Serve(l, http.HandlerFunc(s.handler))
+
+	go func() {
+		ch := <-s.exit
+		ch <- l.Close()
+	}()
+
+	return nil
+}
+
+func (s *RpcServer) Stop() error {
+	ch := make(chan error)
+	s.exit <- ch
+	return <-ch
+}
+
+func NewRpcServer(address string) *RpcServer {
+	return &RpcServer{
+		rpc:     rpc.NewServer(),
+		address: address,
+		exit:    make(chan chan error),
+	}
+}

server/server.go

@@ -0,0 +1,112 @@
+package server
+
+import (
+	"flag"
+	"os"
+	"os/signal"
+	"strconv"
+	"strings"
+	"syscall"
+
+	"code.google.com/p/go-uuid/uuid"
+	"github.com/asim/go-micro/registry"
+	"github.com/asim/go-micro/store"
+	log "github.com/cihub/seelog"
+)
+
+type Server interface {
+	Address() string
+	Init() error
+	NewReceiver(interface{}) Receiver
+	NewNamedReceiver(string, interface{}) Receiver
+	Register(Receiver) error
+	Start() error
+	Stop() error
+}
+
+var (
+	Name          string
+	Id            string
+	DefaultServer Server
+
+	flagRegistry    string
+	flagBindAddress string
+)
+
+func init() {
+	flag.StringVar(&flagRegistry, "registry", "consul", "Registry for discovery. kubernetes, consul, etc")
+	flag.StringVar(&flagBindAddress, "bind_address", ":0", "Bind address for the server. 127.0.0.1:8080")
+}
+
+func Init() error {
+	flag.Parse()
+
+	switch flagRegistry {
+	case "kubernetes":
+		registry.DefaultRegistry = registry.NewKubernetesRegistry()
+		store.DefaultStore = store.NewMemcacheStore()
+	}
+
+	if len(Name) == 0 {
+		Name = "go-server"
+	}
+
+	if len(Id) == 0 {
+		Id = Name + "-" + uuid.NewUUID().String()
+	}
+
+	if DefaultServer == nil {
+		DefaultServer = NewRpcServer(flagBindAddress)
+	}
+
+	return DefaultServer.Init()
+}
+
+func NewReceiver(handler interface{}) Receiver {
+	return DefaultServer.NewReceiver(handler)
+}
+
+func NewNamedReceiver(path string, handler interface{}) Receiver {
+	return DefaultServer.NewNamedReceiver(path, handler)
+}
+
+func Register(r Receiver) error {
+	return DefaultServer.Register(r)
+}
+
+func Run() error {
+	if err := Start(); err != nil {
+		return err
+	}
+
+	// parse address for host, port
+	parts := strings.Split(DefaultServer.Address(), ":")
+	host := strings.Join(parts[:len(parts)-1], ":")
+	port, _ := strconv.Atoi(parts[len(parts)-1])
+
+	// register service
+	node := registry.NewNode(Id, host, port)
+	service := registry.NewService(Name, node)
+
+	log.Debugf("Registering %s", node.Id())
+	registry.Register(service)
+
+	ch := make(chan os.Signal, 1)
+	signal.Notify(ch, syscall.SIGTERM, syscall.SIGINT, syscall.SIGKILL)
+	log.Debugf("Received signal %s", <-ch)
+
+	log.Debugf("Deregistering %s", node.Id())
+	registry.Deregister(service)
+
+	return Stop()
+}
+
+func Start() error {
+	log.Debugf("Starting server %s id %s", Name, Id)
+	return DefaultServer.Start()
+}
+
+func Stop() error {
+	log.Debugf("Stopping server")
+	return DefaultServer.Stop()
+}

server/server_context.go

@@ -0,0 +1,120 @@
+package server
+
+import (
+	"net/http"
+	"sync"
+
+	"github.com/asim/go-micro/client"
+	log "github.com/cihub/seelog"
+)
+
+var ctxs = struct {
+	sync.Mutex
+	m map[*http.Request]*serverContext
+}{
+	m: make(map[*http.Request]*serverContext),
+}
+
+// A server context interface
+type Context interface {
+	Request() Request                                           // the request made to the server
+	Headers() Headers                                           // the response headers
+	NewRequest(string, string, interface{}) client.Request      // a new scoped client request
+	NewProtoRequest(string, string, interface{}) client.Request // a new scoped client request
+	NewJsonRequest(string, string, interface{}) client.Request  // a new scoped client request
+}
+
+// context represents the context of an in-flight HTTP request.
+// It implements the appengine.Context and http.ResponseWriter interfaces.
+type serverContext struct {
+	req       *serverRequest
+	outCode   int
+	outHeader http.Header
+	outBody   []byte
+}
+
+// Copied from $GOROOT/src/pkg/net/http/transfer.go. Some response status
+// codes do not permit a response body (nor response entity headers such as
+// Content-Length, Content-Type, etc).
+func bodyAllowedForStatus(status int) bool {
+	switch {
+	case status >= 100 && status <= 199:
+		return false
+	case status == 204:
+		return false
+	case status == 304:
+		return false
+	}
+	return true
+}
+
+func getServerContext(req *http.Request) *serverContext {
+	ctxs.Lock()
+	c := ctxs.m[req]
+	ctxs.Unlock()
+
+	if c == nil {
+		// Someone passed in an http.Request that is not in-flight.
+		panic("NewContext passed an unknown http.Request")
+	}
+	return c
+}
+
+func (c *serverContext) NewRequest(service, method string, request interface{}) client.Request {
+	req := client.NewRequest(service, method, request)
+	// TODO: set headers and scope
+	req.Headers().Set("X-User-Session", c.Request().Session("X-User-Session"))
+	return req
+}
+
+func (c *serverContext) NewProtoRequest(service, method string, request interface{}) client.Request {
+	req := client.NewProtoRequest(service, method, request)
+	// TODO: set headers and scope
+	req.Headers().Set("X-User-Session", c.Request().Session("X-User-Session"))
+	return req
+}
+
+func (c *serverContext) NewJsonRequest(service, method string, request interface{}) client.Request {
+	req := client.NewJsonRequest(service, method, request)
+	// TODO: set headers and scope
+	req.Headers().Set("X-User-Session", c.Request().Session("X-User-Session"))
+	return req
+}
+
+// The response headers
+func (c *serverContext) Headers() Headers {
+	return c.outHeader
+}
+
+// The response headers
+func (c *serverContext) Header() http.Header {
+	return c.outHeader
+}
+
+// The request made to the server
+func (c *serverContext) Request() Request {
+	return c.req
+}
+
+func (c *serverContext) Write(b []byte) (int, error) {
+	if c.outCode == 0 {
+		c.WriteHeader(http.StatusOK)
+	}
+	if len(b) > 0 && !bodyAllowedForStatus(c.outCode) {
+		return 0, http.ErrBodyNotAllowed
+	}
+	c.outBody = append(c.outBody, b...)
+	return len(b), nil
+}
+
+func (c *serverContext) WriteHeader(code int) {
+	if c.outCode != 0 {
+		log.Errorf("WriteHeader called multiple times on request.")
+		return
+	}
+	c.outCode = code
+}
+
+func GetContext(r *http.Request) *serverContext {
+	return getServerContext(r)
+}

server/server_request.go

@@ -0,0 +1,25 @@
+package server
+
+import (
+	"net/http"
+)
+
+type serverRequest struct {
+	req *http.Request
+}
+
+func (s *serverRequest) Headers() Headers {
+	return s.req.Header
+}
+
+func (s *serverRequest) Session(name string) string {
+	if sess := s.Headers().Get(name); len(sess) > 0 {
+		return sess
+	}
+
+	c, err := s.req.Cookie(name)
+	if err != nil {
+		return ""
+	}
+	return c.Value
+}