services/ws/router.go

   1 package main
   2
   3 import (
   4         "encoding/json"
   5         "log"
   6         "net/http"
   7         "sync"
   8         "time"
   9
  10         "git.curoverse.com/arvados.git/sdk/go/arvados"
  11         "golang.org/x/net/websocket"
  12 )
  13
  14 type router struct {
  15         Config *Config
  16
  17         eventSource eventSource
  18         mux         *http.ServeMux
  19         setupOnce   sync.Once
  20 }
  21
  22 func (rtr *router) setup() {
  23         rtr.mux = http.NewServeMux()
  24         rtr.mux.Handle("/websocket", rtr.makeServer(NewSessionV0))
  25         rtr.mux.Handle("/arvados/v1/events.ws", rtr.makeServer(NewSessionV1))
  26 }
  27
  28 func (rtr *router) makeServer(newSession func(wsConn, arvados.Client) (session, error)) *websocket.Server {
  29         handler := &handler{
  30                 Client:      rtr.Config.Client,
  31                 PingTimeout: rtr.Config.PingTimeout.Duration(),
  32                 QueueSize:   rtr.Config.ClientEventQueue,
  33                 NewSession:  newSession,
  34         }
  35         return &websocket.Server{
  36                 Handshake: func(c *websocket.Config, r *http.Request) error {
  37                         return nil
  38                 },
  39                 Handler: websocket.Handler(func(ws *websocket.Conn) {
  40                         sink := rtr.eventSource.NewSink()
  41                         handler.Handle(ws, sink.Channel())
  42                         sink.Stop()
  43                         ws.Close()
  44                 }),
  45         }
  46 }
  47
  48 func (rtr *router) ServeHTTP(resp http.ResponseWriter, req *http.Request) {
  49         rtr.setupOnce.Do(rtr.setup)
  50         t0 := time.Now()
  51         reqLog(map[string]interface{}{
  52                 "Connect":         req.RemoteAddr,
  53                 "RemoteAddr":      req.RemoteAddr,
  54                 "X-Forwarded-For": req.Header.Get("X-Forwarded-For"),
  55                 "Time":            t0.UTC(),
  56         })
  57         rtr.mux.ServeHTTP(resp, req)
  58         t1 := time.Now()
  59         reqLog(map[string]interface{}{
  60                 "Disconnect":      req.RemoteAddr,
  61                 "RemoteAddr":      req.RemoteAddr,
  62                 "X-Forwarded-For": req.Header.Get("X-Forwarded-For"),
  63                 "Time":            t1.UTC(),
  64                 "Elapsed":         time.Now().Sub(t0).Seconds(),
  65         })
  66 }
  67
  68 func reqLog(m map[string]interface{}) {
  69         j, err := json.Marshal(m)
  70         if err != nil {
  71                 log.Fatal(err)
  72         }
  73         log.Print(string(j))
  74 }