Start api revamp

api/database.go

@@ -0,0 +1,359 @@
+package api
+
+import (
+	"bytes"
+	"fmt"
+	"net/http"
+
+	"github.com/gorilla/websocket"
+	"github.com/tevino/abool"
+
+	"github.com/Safing/portbase/container"
+	"github.com/Safing/portbase/database"
+	"github.com/Safing/portbase/database/query"
+	"github.com/Safing/portbase/database/record"
+	"github.com/Safing/portbase/log"
+)
+
+const (
+	dbMsgTypeOk      = "ok"
+	dbMsgTypeError   = "error"
+	dbMsgTypeDone    = "done"
+	dbMsgTypeSuccess = "success"
+	dbMsgTypeUpd     = "upd"
+	dbMsgTypeNew     = "new"
+	dbMsgTypeDelete  = "delete"
+	dbMsgTypeWarning = "warning"
+)
+
+// DatabaseAPI is a database API instance.
+type DatabaseAPI struct {
+	conn      *websocket.Conn
+	sendQueue chan []byte
+	subs      map[string]*database.Subscription
+
+	shutdownSignal chan struct{}
+	shuttingDown   *abool.AtomicBool
+	db             *database.Interface
+}
+
+func allowAnyOrigin(r *http.Request) bool {
+	return true
+}
+
+func startDatabaseAPI(w http.ResponseWriter, r *http.Request) {
+
+	upgrader := websocket.Upgrader{
+		CheckOrigin:     allowAnyOrigin,
+		ReadBufferSize:  1024,
+		WriteBufferSize: 65536,
+	}
+	wsConn, err := upgrader.Upgrade(w, r, nil)
+	if err != nil {
+		errMsg := fmt.Sprintf("could not upgrade to websocket: %s", err)
+		log.Error(errMsg)
+		http.Error(w, errMsg, 400)
+		return
+	}
+
+	new := &DatabaseAPI{
+		conn:           wsConn,
+		sendQueue:      make(chan []byte, 100),
+		subs:           make(map[string]*database.Subscription),
+		shutdownSignal: make(chan struct{}),
+		shuttingDown:   abool.NewBool(false),
+		db:             database.NewInterface(nil),
+	}
+
+	go new.handler()
+	go new.writer()
+}
+
+func (api *DatabaseAPI) handler() {
+
+	// 123|get|<key>
+	//    123|ok|<key>|<data>
+	//    123|error|<message>
+	// 124|query|<query>
+	//    124|ok|<key>|<data>
+	//    124|done
+	//    124|error|<message>
+	// 125|sub|<query>
+	//    125|upd|<key>|<data>
+	//    125|new|<key>|<data>
+	//    125|delete|<key>|<data>
+	//    125|warning|<message> // does not cancel the subscription
+	// 127|qsub|<query>
+	//    127|ok|<key>|<data>
+	//    127|done
+	//    127|error|<message>
+	//    127|upd|<key>|<data>
+	//    127|new|<key>|<data>
+	//    127|delete|<key>|<data>
+	//    127|warning|<message> // does not cancel the subscription
+
+	// 128|create|<key>|<data>
+	//    128|success
+	//    128|error|<message>
+	// 129|update|<key>|<data>
+	//    129|success
+	//    129|error|<message>
+	// 130|insert|<key>|<data>
+	//    130|success
+	//    130|error|<message>
+
+	for {
+
+		_, msg, err := api.conn.ReadMessage()
+		if err != nil {
+			if !api.shuttingDown.IsSet() {
+				api.shutdown()
+				if !websocket.IsCloseError(err, websocket.CloseNormalClosure, websocket.CloseGoingAway) {
+					log.Warningf("api: websocket write error: %s", err)
+				}
+			}
+			return
+		}
+
+		parts := bytes.SplitN(msg, []byte("|"), 2)
+		if len(parts) != 3 {
+			api.send(nil, dbMsgTypeError, []byte("bad request: malformed message"))
+			continue
+		}
+
+		switch string(parts[1]) {
+		case "get":
+			// 123|get|<key>
+			go api.handleGet(parts[0], string(parts[2]))
+		case "query":
+			// 124|query|<query>
+			go api.handleQuery(parts[0], string(parts[2]))
+		case "sub":
+			// 125|sub|<query>
+			go api.handleSub(parts[0], string(parts[2]))
+		case "qsub":
+			// 127|qsub|<query>
+			go api.handleQsub(parts[0], string(parts[2]))
+		case "create", "update", "insert":
+
+			// split key and payload
+			dataParts := bytes.SplitN(parts[2], []byte("|"), 1)
+			if len(dataParts) != 2 {
+				api.send(nil, dbMsgTypeError, []byte("bad request: malformed message"))
+				continue
+			}
+
+			switch string(parts[1]) {
+			case "create":
+				// 128|create|<key>|<data>
+				go api.handleCreate(parts[0], string(dataParts[0]), dataParts[1])
+			case "update":
+				// 129|update|<key>|<data>
+				go api.handleUpdate(parts[0], string(dataParts[0]), dataParts[1])
+			case "insert":
+				// 130|insert|<key>|<data>
+				go api.handleInsert(parts[0], string(dataParts[0]), dataParts[1])
+			}
+
+		default:
+			api.send(parts[0], dbMsgTypeError, []byte("bad request: unknown method"))
+		}
+	}
+}
+
+func (api *DatabaseAPI) writer() {
+	var data []byte
+	var err error
+
+	for {
+		data = nil
+
+		select {
+		// prioritize direct writes
+		case data = <-api.sendQueue:
+			if data == nil || len(data) == 0 {
+				api.shutdown()
+				return
+			}
+		case <-api.shutdownSignal:
+			return
+		}
+
+		// log.Tracef("api: sending %s", string(*msg))
+		err = api.conn.WriteMessage(websocket.BinaryMessage, data)
+		if err != nil {
+			if !api.shuttingDown.IsSet() {
+				api.shutdown()
+				if !websocket.IsCloseError(err, websocket.CloseNormalClosure, websocket.CloseGoingAway) {
+					log.Warningf("api: websocket write error: %s", err)
+				}
+			}
+			return
+		}
+
+	}
+}
+
+func (api *DatabaseAPI) send(opID []byte, msgType string, data []byte) {
+	c := container.New(opID)
+	c.Append([]byte(fmt.Sprintf("|%s|", msgType)))
+	c.Append(data)
+	api.sendQueue <- c.CompileData()
+}
+
+func (api *DatabaseAPI) handleGet(opID []byte, key string) {
+	// 123|get|<key>
+	//    123|ok|<key>|<data>
+	//    123|error|<message>
+
+	var data []byte
+
+	r, err := api.db.Get(key)
+	if err == nil {
+		data, err = r.Marshal(r, record.JSON)
+	} else {
+		api.send(opID, dbMsgTypeError, []byte(err.Error()))
+		return
+	}
+	api.send(opID, dbMsgTypeOk, data)
+}
+
+func (api *DatabaseAPI) handleQuery(opID []byte, queryText string) {
+	// 124|query|<query>
+	//    124|ok|<key>|<data>
+	//    124|done
+	//    124|warning|<message>
+	//    124|error|<message>
+
+	var err error
+
+	q, err := query.ParseQuery(queryText)
+	if err != nil {
+		api.send(opID, dbMsgTypeError, []byte(err.Error()))
+		return
+	}
+
+	api.processQuery(opID, q)
+}
+
+func (api *DatabaseAPI) processQuery(opID []byte, q *query.Query) (ok bool) {
+	it, err := api.db.Query(q)
+	if err != nil {
+		api.send(opID, dbMsgTypeError, []byte(err.Error()))
+		return false
+	}
+
+	for r := range it.Next {
+		data, err := r.Marshal(r, record.JSON)
+		if err != nil {
+			api.send(opID, dbMsgTypeWarning, []byte(err.Error()))
+		}
+		api.send(opID, dbMsgTypeOk, data)
+	}
+	if it.Error != nil {
+		api.send(opID, dbMsgTypeError, []byte(err.Error()))
+		return false
+	}
+
+	api.send(opID, dbMsgTypeDone, nil)
+	return true
+}
+
+// func (api *DatabaseAPI) runQuery()
+
+func (api *DatabaseAPI) handleSub(opID []byte, queryText string) {
+	// 125|sub|<query>
+	//    125|upd|<key>|<data>
+	//    125|new|<key>|<data>
+	//    125|delete|<key>
+	//    125|warning|<message> // does not cancel the subscription
+	var err error
+
+	q, err := query.ParseQuery(queryText)
+	if err != nil {
+		api.send(opID, dbMsgTypeError, []byte(err.Error()))
+		return
+	}
+
+	sub, ok := api.registerSub(opID, q)
+	if !ok {
+		return
+	}
+	api.processSub(opID, sub)
+}
+
+func (api *DatabaseAPI) registerSub(opID []byte, q *query.Query) (sub *database.Subscription, ok bool) {
+	var err error
+	sub, err = api.db.Subscribe(q)
+	if err != nil {
+		api.send(opID, dbMsgTypeError, []byte(err.Error()))
+		return nil, false
+	}
+	return sub, true
+}
+
+func (api *DatabaseAPI) processSub(opID []byte, sub *database.Subscription) {
+	for r := range sub.Feed {
+		data, err := r.Marshal(r, record.JSON)
+		if err != nil {
+			api.send(opID, dbMsgTypeWarning, []byte(err.Error()))
+		}
+		// TODO: use upd, new and delete msgTypes
+		api.send(opID, dbMsgTypeOk, data)
+	}
+	if sub.Err != nil {
+		api.send(opID, dbMsgTypeError, []byte(sub.Err.Error()))
+	}
+}
+
+func (api *DatabaseAPI) handleQsub(opID []byte, queryText string) {
+	// 127|qsub|<query>
+	//    127|ok|<key>|<data>
+	//    127|done
+	//    127|error|<message>
+	//    127|upd|<key>|<data>
+	//    127|new|<key>|<data>
+	//    127|delete|<key>
+	//    127|warning|<message> // does not cancel the subscription
+
+	var err error
+
+	q, err := query.ParseQuery(queryText)
+	if err != nil {
+		api.send(opID, dbMsgTypeError, []byte(err.Error()))
+		return
+	}
+
+	sub, ok := api.registerSub(opID, q)
+	if !ok {
+		return
+	}
+	ok = api.processQuery(opID, q)
+	if !ok {
+		return
+	}
+	api.processSub(opID, sub)
+}
+
+func (api *DatabaseAPI) handleCreate(opID []byte, key string, data []byte) {
+	// 128|create|<key>|<data>
+	//    128|success
+	//    128|error|<message>
+}
+func (api *DatabaseAPI) handleUpdate(opID []byte, key string, data []byte) {
+	// 129|update|<key>|<data>
+	//    129|success
+	//    129|error|<message>
+}
+func (api *DatabaseAPI) handleInsert(opID []byte, key string, data []byte) {
+	// 130|insert|<key>|<data>
+	//    130|success
+	//    130|error|<message>
+}
+
+func (api *DatabaseAPI) shutdown() {
+	if api.shuttingDown.SetToIf(false, true) {
+		close(api.shutdownSignal)
+		api.conn.Close()
+	}
+}

api/old/router.go

@@ -0,0 +1,28 @@
+// Copyright Safing ICS Technologies GmbH. Use of this source code is governed by the AGPL license that can be found in the LICENSE file.
+
+package api
+
+import (
+	"net/http"
+
+	"github.com/gorilla/mux"
+)
+
+func NewRouter() *mux.Router {
+	router := mux.NewRouter().StrictSlash(true)
+
+	for _, route := range routes {
+		var handler http.Handler
+
+		handler = route.Handler
+		handler = Logger(handler, route.Name)
+
+		router.
+			Methods(route.Method).
+			PathPrefix(route.Path).
+			Name(route.Name).
+			Handler(handler)
+	}
+
+	return router
+}

api/router.go

@@ -1,5 +1,3 @@
-// Copyright Safing ICS Technologies GmbH. Use of this source code is governed by the AGPL license that can be found in the LICENSE file.
-
 package api
 
 import (
@@ -8,21 +6,24 @@ import (
 	"github.com/gorilla/mux"
 )
 
-func NewRouter() *mux.Router {
+var (
-	router := mux.NewRouter().StrictSlash(true)
+	additionalRoutes map[string]func(arg1 http.ResponseWriter, arg2 *http.Request)
+)
 
-	for _, route := range routes {
+func RegisterAdditionalRoute(path string, handleFunc func(arg1 http.ResponseWriter, arg2 *http.Request)) {
-		var handler http.Handler
+	if additionalRoutes == nil {
+		additionalRoutes = make(map[string]func(arg1 http.ResponseWriter, arg2 *http.Request))
+	}
+	additionalRoutes[path] = handleFunc
+}
 
-		handler = route.Handler
+func Serve() {
-		handler = Logger(handler, route.Name)
 
-		router.
+	router := mux.NewRouter()
-			Methods(route.Method).
+	router.HandleFunc("/api/database/v1", startDatabaseAPI)
-			PathPrefix(route.Path).
+
-			Name(route.Name).
+	for path, handleFunc := range additionalRoutes {
-			Handler(handler)
+		router.HandleFunc(path, handleFunc)
 	}
 
-	return router
 }

api/security.go

@@ -0,0 +1 @@
+package api

api/websocket.go

@@ -0,0 +1 @@
+package api

database/controller.go

@@ -129,7 +129,7 @@ func (c *Controller) Put(r record.Record) (err error) {
 
 	// process subscriptions
 	for _, sub := range c.subscriptions {
-		if sub.q.Matches(r) {
+		if r.Meta().CheckPermission(sub.local, sub.internal) && sub.q.Matches(r) {
 			select {
 			case sub.Feed <- r:
 			default:

database/database_test.go

@@ -36,15 +36,15 @@ func testDatabase(t *testing.T, storageType string) {
 		t.Fatal(err)
 	}
 
+	// interface
+	db := NewInterface(nil)
+
 	// sub
-	sub, err := Subscribe(q.New(dbName).MustBeValid())
+	sub, err := db.Subscribe(q.New(dbName).MustBeValid())
 	if err != nil {
 		t.Fatal(err)
 	}
 
-	// interface
-	db := NewInterface(nil)
-
 	A := NewExample(makeKey(dbName, "A"), "Herbert", 411)
 	err = A.Save()
 	if err != nil {

database/interface.go

@@ -228,6 +228,11 @@ func (i *Interface) Delete(key string) error {
 
 // Query executes the given query on the database.
 func (i *Interface) Query(q *query.Query) (*iterator.Iterator, error) {
+	_, err := q.Check()
+	if err != nil {
+		return nil, err
+	}
+
 	db, err := getController(q.DatabaseName())
 	if err != nil {
 		return nil, err
@@ -235,3 +240,30 @@ func (i *Interface) Query(q *query.Query) (*iterator.Iterator, error) {
 
 	return db.Query(q, i.options.Local, i.options.Internal)
 }
+
+// Subscribe subscribes to updates matching the given query.
+func (i *Interface) Subscribe(q *query.Query) (*Subscription, error) {
+	_, err := q.Check()
+	if err != nil {
+		return nil, err
+	}
+
+	c, err := getController(q.DatabaseName())
+	if err != nil {
+		return nil, err
+	}
+
+	c.readLock.Lock()
+	defer c.readLock.Unlock()
+	c.writeLock.Lock()
+	defer c.writeLock.Unlock()
+
+	sub := &Subscription{
+		q:        q,
+		local:    i.options.Local,
+		internal: i.options.Internal,
+		Feed:     make(chan record.Record, 100),
+	}
+	c.subscriptions = append(c.subscriptions, sub)
+	return sub, nil
+}

database/subscription.go

@@ -7,36 +7,15 @@ import (
 
 // Subscription is a database subscription for updates.
 type Subscription struct {
-	q    *query.Query
+	q        *query.Query
+	local    bool
+	internal bool
+	canceled bool
+
 	Feed chan record.Record
 	Err  error
 }
 
-// Subscribe subscribes to updates matching the given query.
-func Subscribe(q *query.Query) (*Subscription, error) {
-	_, err := q.Check()
-	if err != nil {
-		return nil, err
-	}
-
-	c, err := getController(q.DatabaseName())
-	if err != nil {
-		return nil, err
-	}
-
-	c.readLock.Lock()
-	defer c.readLock.Unlock()
-	c.writeLock.Lock()
-	defer c.writeLock.Unlock()
-
-	sub := &Subscription{
-		q:    q,
-		Feed: make(chan record.Record, 100),
-	}
-	c.subscriptions = append(c.subscriptions, sub)
-	return sub, nil
-}
-
 // Cancel cancels the subscription.
 func (s *Subscription) Cancel() error {
 	c, err := getController(s.q.DatabaseName())
@@ -49,6 +28,12 @@ func (s *Subscription) Cancel() error {
 	c.writeLock.Lock()
 	defer c.writeLock.Unlock()
 
+	if s.canceled {
+		return nil
+	}
+	s.canceled = true
+	close(s.Feed)
+
 	for key, sub := range c.subscriptions {
 		if sub.q == s.q {
 			c.subscriptions = append(c.subscriptions[:key], c.subscriptions[key+1:]...)