services/ws/session_v0.go

   1 package main
   2
   3 import (
   4         "encoding/json"
   5         "errors"
   6         "log"
   7         "sync"
   8
   9         "git.curoverse.com/arvados.git/sdk/go/arvados"
  10 )
  11
  12 var (
  13         errQueueFull   = errors.New("client queue full")
  14         errFrameTooBig = errors.New("frame too big")
  15 )
  16
  17 type sessionV0 struct {
  18         ws          wsConn
  19         permChecker permChecker
  20         subscribed  map[string]bool
  21         mtx         sync.Mutex
  22         setupOnce   sync.Once
  23 }
  24
  25 func NewSessionV0(ws wsConn, ac arvados.Client) (session, error) {
  26         sess := &sessionV0{
  27                 ws:          ws,
  28                 permChecker: NewPermChecker(ac),
  29                 subscribed:  make(map[string]bool),
  30         }
  31
  32         err := ws.Request().ParseForm()
  33         if err != nil {
  34                 log.Printf("%s ParseForm: %s", ws.Request().RemoteAddr, err)
  35                 return nil, err
  36         }
  37         token := ws.Request().Form.Get("api_token")
  38         sess.permChecker.SetToken(token)
  39         sess.debugLogf("token = %+q", token)
  40
  41         return sess, nil
  42 }
  43
  44 func (sess *sessionV0) debugLogf(s string, args ...interface{}) {
  45         args = append([]interface{}{sess.ws.Request().RemoteAddr}, args...)
  46         debugLogf("%s "+s, args...)
  47 }
  48
  49 func (sess *sessionV0) Receive(msg map[string]interface{}) {
  50         sess.debugLogf("received message: %+v", msg)
  51         sess.debugLogf("subscribing to *")
  52         sess.subscribed["*"] = true
  53 }
  54
  55 func (sess *sessionV0) EventMessage(e *event) ([]byte, error) {
  56         detail := e.Detail()
  57         if detail == nil {
  58                 return nil, nil
  59         }
  60
  61         ok, err := sess.permChecker.Check(detail.UUID)
  62         if err != nil || !ok {
  63                 return nil, err
  64         }
  65
  66         return json.Marshal(map[string]interface{}{
  67                 "msgID":             e.Serial,
  68                 "id":                detail.ID,
  69                 "uuid":              detail.UUID,
  70                 "object_uuid":       detail.ObjectUUID,
  71                 "object_owner_uuid": detail.ObjectOwnerUUID,
  72                 "event_type":        detail.EventType,
  73         })
  74 }
  75
  76 func (sess *sessionV0) Filter(e *event) bool {
  77         detail := e.Detail()
  78         sess.mtx.Lock()
  79         defer sess.mtx.Unlock()
  80         switch {
  81         case sess.subscribed["*"]:
  82                 return true
  83         case detail == nil:
  84                 return false
  85         case sess.subscribed[detail.ObjectUUID]:
  86                 return true
  87         case sess.subscribed[detail.ObjectOwnerUUID]:
  88                 return true
  89         default:
  90                 return false
  91         }
  92 }