Merge lp:~pedronis/ubuntu-push/store-flex into lp:ubuntu-push

[Samuele Pedroni]

add some flexibility in how the requests get the store; log unexpected store related errors

add some flexibility in how the requests get the store

log unexpected store related errors

[Guillermo Gonzalez (verterok)]

looks good.

review: Approve

=== modified file 'server/api/handlers.go'
--- server/api/handlers.go	2014-02-13 18:49:31 +0000
+++ server/api/handlers.go	2014-02-18 14:20:50 +0000
@@ -118,6 +118,11 @@
 		internalError,
 		"Unknown error",
 	}
+	ErrStoreUnavailable = &APIError{
+		http.StatusServiceUnavailable,
+		unavailable,
+		"Message store unavailable",
+	}
 	ErrCouldNotStoreNotification = &APIError{
 		http.StatusServiceUnavailable,
 		unavailable,
@@ -205,21 +210,38 @@
 	return expire, nil
 }
 
-// state holds the interfaces to delegate to serving requests
-type state struct {
-	store  store.PendingStore
-	broker broker.BrokerSending
-	logger logger.Logger
-}
-
-type BroadcastHandler state
-
-func (h *BroadcastHandler) doBroadcast(bcast *Broadcast) *APIError {
+type StoreForRequest func(w http.ResponseWriter, request *http.Request) (store.PendingStore, error)
+
+// context holds the interfaces to delegate to serving requests
+type context struct {
+	storeForRequest StoreForRequest
+	broker          broker.BrokerSending
+	logger          logger.Logger
+}
+
+func (ctx *context) getStore(w http.ResponseWriter, request *http.Request) (store.PendingStore, *APIError) {
+	sto, err := ctx.storeForRequest(w, request)
+	if err != nil {
+		apiErr, ok := err.(*APIError)
+		if ok {
+			return nil, apiErr
+		}
+		ctx.logger.Errorf("failed to get store: %v", err)
+		return nil, ErrUnknown
+	}
+	return sto, nil
+}
+
+type BroadcastHandler struct {
+	*context
+}
+
+func (h *BroadcastHandler) doBroadcast(sto store.PendingStore, bcast *Broadcast) *APIError {
 	expire, apiErr := checkBroadcast(bcast)
 	if apiErr != nil {
 		return apiErr
 	}
-	chanId, err := h.store.GetInternalChannelId(bcast.Channel)
+	chanId, err := sto.GetInternalChannelId(bcast.Channel)
 	if err != nil {
 		switch err {
 		case store.ErrUnknownChannel:
@@ -228,9 +250,9 @@
 			return ErrUnknown
 		}
 	}
-	err = h.store.AppendToChannel(chanId, bcast.Data, expire)
+	err = sto.AppendToChannel(chanId, bcast.Data, expire)
 	if err != nil {
-		// assume this for now
+		h.logger.Errorf("could not store notification: %v", err)
 		return ErrCouldNotStoreNotification
 	}
 
@@ -239,24 +261,33 @@
 }
 
 func (h *BroadcastHandler) ServeHTTP(writer http.ResponseWriter, request *http.Request) {
+	var apiErr *APIError
+	defer func() {
+		if apiErr != nil {
+			respondError(writer, apiErr)
+		}
+	}()
+
 	body, apiErr := readBody(request)
+	if apiErr != nil {
+		return
+	}
 
+	sto, apiErr := h.getStore(writer, request)
 	if apiErr != nil {
-		respondError(writer, apiErr)
 		return
 	}
+	defer sto.Close()
 
 	broadcast := &Broadcast{}
 	err := json.Unmarshal(body, broadcast)
-
 	if err != nil {
-		respondError(writer, ErrMalformedJSONObject)
+		apiErr = ErrMalformedJSONObject
 		return
 	}
 
-	apiErr = h.doBroadcast(broadcast)
+	apiErr = h.doBroadcast(sto, broadcast)
 	if apiErr != nil {
-		respondError(writer, apiErr)
 		return
 	}
 
@@ -265,12 +296,13 @@
 }
 
 // MakeHandlersMux makes a handler that dispatches for the various API endpoints.
-func MakeHandlersMux(store store.PendingStore, broker broker.BrokerSending, logger logger.Logger) *http.ServeMux {
+func MakeHandlersMux(storeForRequest StoreForRequest, broker broker.BrokerSending, logger logger.Logger) *http.ServeMux {
+	ctx := &context{
+		storeForRequest: storeForRequest,
+		broker:          broker,
+		logger:          logger,
+	}
 	mux := http.NewServeMux()
-	mux.Handle("/broadcast", &BroadcastHandler{
-		store:  store,
-		broker: broker,
-		logger: logger,
-	})
+	mux.Handle("/broadcast", &BroadcastHandler{context: ctx})
 	return mux
 }

=== modified file 'server/api/handlers_test.go'
--- server/api/handlers_test.go	2014-02-10 23:29:53 +0000
+++ server/api/handlers_test.go	2014-02-18 14:20:50 +0000
@@ -31,6 +31,7 @@
 	. "launchpad.net/gocheck"
 
 	"launchpad.net/ubuntu-push/server/store"
+	helpers "launchpad.net/ubuntu-push/testing"
 )
 
 func TestHandlers(t *testing.T) { TestingT(t) }
@@ -40,12 +41,14 @@
 	json            string
 	client          *http.Client
 	c               *C
+	testlog         *helpers.TestLogger
 }
 
 var _ = Suite(&handlersSuite{})
 
 func (s *handlersSuite) SetUpTest(c *C) {
 	s.client = &http.Client{}
+	s.testlog = helpers.NewTestLogger(c, "error")
 }
 
 func (s *handlersSuite) TestAPIError(c *C) {
@@ -66,6 +69,23 @@
 	c.Check(err, Equals, ErrCouldNotReadBody)
 }
 
+func (s *handlersSuite) TestGetStore(c *C) {
+	ctx := &context{storeForRequest: func(w http.ResponseWriter, r *http.Request) (store.PendingStore, error) {
+		return nil, ErrStoreUnavailable
+	}}
+	sto, apiErr := ctx.getStore(nil, nil)
+	c.Check(sto, IsNil)
+	c.Check(apiErr, Equals, ErrStoreUnavailable)
+
+	ctx = &context{storeForRequest: func(w http.ResponseWriter, r *http.Request) (store.PendingStore, error) {
+		return nil, errors.New("something else")
+	}, logger: s.testlog}
+	sto, apiErr = ctx.getStore(nil, nil)
+	c.Check(sto, IsNil)
+	c.Check(apiErr, Equals, ErrUnknown)
+	c.Check(s.testlog.Captured(), Equals, "ERROR failed to get store: something else\n")
+}
+
 var future = time.Now().Add(4 * time.Hour).Format(time.RFC3339)
 
 func (s *handlersSuite) TestCheckBroadcast(c *C) {
@@ -122,9 +142,9 @@
 func (s *handlersSuite) TestDoBroadcast(c *C) {
 	sto := store.NewInMemoryPendingStore()
 	bsend := &checkBrokerSending{store: sto}
-	bh := &BroadcastHandler{sto, bsend, nil}
+	bh := &BroadcastHandler{&context{nil, bsend, nil}}
 	payload := json.RawMessage(`{"a": 1}`)
-	apiErr := bh.doBroadcast(&Broadcast{
+	apiErr := bh.doBroadcast(sto, &Broadcast{
 		Channel:  "system",
 		ExpireOn: future,
 		Data:     payload,
@@ -138,8 +158,8 @@
 
 func (s *handlersSuite) TestDoBroadcastUnknownChannel(c *C) {
 	sto := store.NewInMemoryPendingStore()
-	bh := &BroadcastHandler{sto, nil, nil}
-	apiErr := bh.doBroadcast(&Broadcast{
+	bh := &BroadcastHandler{}
+	apiErr := bh.doBroadcast(sto, &Broadcast{
 		Channel:  "unknown",
 		ExpireOn: future,
 		Data:     json.RawMessage(`{"a": 1}`),
@@ -169,8 +189,8 @@
 			return errors.New("other")
 		},
 	}
-	bh := &BroadcastHandler{sto, nil, nil}
-	apiErr := bh.doBroadcast(&Broadcast{
+	bh := &BroadcastHandler{}
+	apiErr := bh.doBroadcast(sto, &Broadcast{
 		Channel:  "system",
 		ExpireOn: future,
 		Data:     json.RawMessage(`{"a": 1}`),
@@ -188,13 +208,15 @@
 			return err
 		},
 	}
-	bh := &BroadcastHandler{sto, nil, nil}
-	apiErr := bh.doBroadcast(&Broadcast{
+	ctx := &context{logger: s.testlog}
+	bh := &BroadcastHandler{ctx}
+	apiErr := bh.doBroadcast(sto, &Broadcast{
 		Channel:  "system",
 		ExpireOn: future,
 		Data:     json.RawMessage(`{"a": 1}`),
 	})
 	c.Check(apiErr, Equals, ErrCouldNotStoreNotification)
+	c.Check(s.testlog.Captured(), Equals, "ERROR could not store notification: fail\n")
 }
 
 func newPostRequest(path string, message interface{}, server *httptest.Server) *http.Request {
@@ -238,8 +260,11 @@
 
 func (s *handlersSuite) TestRespondsToBasicSystemBroadcast(c *C) {
 	sto := store.NewInMemoryPendingStore()
+	stoForReq := func(http.ResponseWriter, *http.Request) (store.PendingStore, error) {
+		return sto, nil
+	}
 	bsend := testBrokerSending{make(chan store.InternalChannelId, 1)}
-	testServer := httptest.NewServer(MakeHandlersMux(sto, bsend, nil))
+	testServer := httptest.NewServer(MakeHandlersMux(stoForReq, bsend, nil))
 	defer testServer.Close()
 
 	payload := json.RawMessage(`{"foo":"bar"}`)
@@ -268,9 +293,32 @@
 	c.Check(<-bsend.chanId, Equals, store.SystemInternalChannelId)
 }
 
+func (s *handlersSuite) TestStoreUnavailable(c *C) {
+	stoForReq := func(http.ResponseWriter, *http.Request) (store.PendingStore, error) {
+		return nil, ErrStoreUnavailable
+	}
+	testServer := httptest.NewServer(MakeHandlersMux(stoForReq, nil, nil))
+	defer testServer.Close()
+
+	payload := json.RawMessage(`{"foo":"bar"}`)
+
+	request := newPostRequest("/broadcast", &Broadcast{
+		Channel:  "system",
+		ExpireOn: future,
+		Data:     payload,
+	}, testServer)
+
+	response, err := s.client.Do(request)
+	c.Assert(err, IsNil)
+	checkError(c, response, ErrStoreUnavailable)
+}
+
 func (s *handlersSuite) TestFromBroadcastError(c *C) {
 	sto := store.NewInMemoryPendingStore()
-	testServer := httptest.NewServer(MakeHandlersMux(sto, nil, nil))
+	stoForReq := func(http.ResponseWriter, *http.Request) (store.PendingStore, error) {
+		return sto, nil
+	}
+	testServer := httptest.NewServer(MakeHandlersMux(stoForReq, nil, nil))
 	defer testServer.Close()
 
 	payload := json.RawMessage(`{"foo":"bar"}`)
@@ -287,7 +335,11 @@
 }
 
 func (s *handlersSuite) TestMissingData(c *C) {
-	testServer := httptest.NewServer(&BroadcastHandler{})
+	stoForReq := func(http.ResponseWriter, *http.Request) (store.PendingStore, error) {
+		return store.NewInMemoryPendingStore(), nil
+	}
+	ctx := &context{stoForReq, nil, nil}
+	testServer := httptest.NewServer(&BroadcastHandler{ctx})
 	defer testServer.Close()
 
 	packedMessage := []byte(`{"channel": "system"}`)
@@ -304,7 +356,11 @@
 }
 
 func (s *handlersSuite) TestCannotBroadcastMalformedData(c *C) {
-	testServer := httptest.NewServer(&BroadcastHandler{})
+	stoForReq := func(http.ResponseWriter, *http.Request) (store.PendingStore, error) {
+		return store.NewInMemoryPendingStore(), nil
+	}
+	ctx := &context{stoForReq, nil, nil}
+	testServer := httptest.NewServer(&BroadcastHandler{ctx})
 	defer testServer.Close()
 
 	packedMessage := []byte("{some bogus-message: ")

=== modified file 'server/dev/server.go'
--- server/dev/server.go	2014-02-13 19:33:03 +0000
+++ server/dev/server.go	2014-02-18 14:20:50 +0000
@@ -19,6 +19,7 @@
 
 import (
 	"net"
+	"net/http"
 	"os"
 	"path/filepath"
 
@@ -56,7 +57,10 @@
 	broker.Start()
 	defer broker.Stop()
 	// serve the http api
-	mux := api.MakeHandlersMux(sto, broker, logger)
+	storeForRequest := func(http.ResponseWriter, *http.Request) (store.PendingStore, error) {
+		return sto, nil
+	}
+	mux := api.MakeHandlersMux(storeForRequest, broker, logger)
 	handler := api.PanicTo500Handler(mux, logger)
 	go server.HTTPServeRunner(handler, &cfg.HTTPServeParsedConfig)()
 	// listen for device connections

=== modified file 'server/store/inmemory.go'
--- server/store/inmemory.go	2014-02-10 18:43:51 +0000
+++ server/store/inmemory.go	2014-02-18 14:20:50 +0000
@@ -90,5 +90,9 @@
 	return topLevel, res, nil
 }
 
+func (sto *InMemoryPendingStore) Close() {
+	// ignored
+}
+
 // sanity check we implement the interface
-var _ PendingStore = &InMemoryPendingStore{}
+var _ PendingStore = (*InMemoryPendingStore)(nil)

=== modified file 'server/store/store.go'
--- server/store/store.go	2014-02-10 19:53:00 +0000
+++ server/store/store.go	2014-02-18 14:20:50 +0000
@@ -71,4 +71,6 @@
 	// GetChannelSnapshot gets all the current notifications and
 	// current top level in the channel.
 	GetChannelSnapshot(chanId InternalChannelId) (topLevel int64, payloads []json.RawMessage, err error)
+	// Close is to be called when done with the store.
+	Close()
 }