13 "github.com/Sirupsen/logrus"
14 "golang.org/x/net/websocket"
17 type wsConn interface {
19 Request() *http.Request
20 SetReadDeadline(time.Time) error
21 SetWriteDeadline(time.Time) error
26 eventSource eventSource
27 newPermChecker func() permChecker
39 type routerStatus struct {
44 type Statuser interface {
48 type sessionFactory func(wsConn, chan<- interface{}, *sql.DB, permChecker) (session, error)
50 func (rtr *router) setup() {
51 rtr.handler = &handler{
52 PingTimeout: rtr.Config.PingTimeout.Duration(),
53 QueueSize: rtr.Config.ClientEventQueue,
55 rtr.mux = http.NewServeMux()
56 rtr.mux.Handle("/websocket", rtr.makeServer(NewSessionV0))
57 rtr.mux.Handle("/arvados/v1/events.ws", rtr.makeServer(NewSessionV1))
58 rtr.mux.HandleFunc("/status.json", rtr.serveStatus)
61 func (rtr *router) makeServer(newSession sessionFactory) *websocket.Server {
62 return &websocket.Server{
63 Handshake: func(c *websocket.Config, r *http.Request) error {
66 Handler: websocket.Handler(func(ws *websocket.Conn) {
68 log := logger(ws.Request().Context())
71 stats := rtr.handler.Handle(ws, rtr.eventSource,
72 func(ws wsConn, sendq chan<- interface{}) (session, error) {
73 return newSession(ws, sendq, rtr.eventSource.DB(), rtr.newPermChecker())
76 log.WithFields(logrus.Fields{
77 "Elapsed": time.Now().Sub(t0).Seconds(),
85 func (rtr *router) newReqID() string {
87 defer rtr.lastReqMtx.Unlock()
88 id := time.Now().UnixNano()
89 if id <= rtr.lastReqID {
90 id = rtr.lastReqID + 1
92 return strconv.FormatInt(id, 36)
95 func (rtr *router) Status() interface{} {
96 s := map[string]interface{}{
98 "Outgoing": rtr.handler.Status(),
100 if es, ok := rtr.eventSource.(Statuser); ok {
101 s["EventSource"] = es.Status()
106 func (rtr *router) serveStatus(resp http.ResponseWriter, req *http.Request) {
107 rtr.setupOnce.Do(rtr.setup)
108 logger := logger(req.Context())
109 logger.Debug("status")
110 enc := json.NewEncoder(resp)
111 err := enc.Encode(rtr.Status())
113 logger.WithError(err).Error("status encode failed")
117 func (rtr *router) ServeHTTP(resp http.ResponseWriter, req *http.Request) {
118 rtr.setupOnce.Do(rtr.setup)
119 atomic.AddInt64(&rtr.status.ReqsReceived, 1)
120 atomic.AddInt64(&rtr.status.ReqsActive, 1)
121 defer atomic.AddInt64(&rtr.status.ReqsActive, -1)
123 logger := logger(req.Context()).
124 WithField("RequestID", rtr.newReqID())
125 ctx := contextWithLogger(req.Context(), logger)
126 req = req.WithContext(ctx)
127 logger.WithFields(logrus.Fields{
128 "RemoteAddr": req.RemoteAddr,
129 "X-Forwarded-For": req.Header.Get("X-Forwarded-For"),
130 }).Info("accept request")
131 rtr.mux.ServeHTTP(resp, req)