9f8659abd6
This change adds a new websocket endpoint for database events. The events
endpoint allows clients to stream events as they happen in GARM. Events
are defined as a structure containning the event type (create, update, delete),
the database entity involved (instances, pools, repos, etc) and the payload
consisting of the object involved in the event. The payload translates
to the types normally returned by the API and can be deserialized as one
of the types present in the params package.
The events endpoint is a websocket endpoint and it accepts filters as
a simple json send over the websocket connection. The filters allows the
user to specify which entities are of interest, and which operations should
be returned. For example, you may be interested in changes made to pools
or runners, in which case you could create a filter that only returns
update operations for pools. Or update and delete operations.
The filters can be defined as:
{
"filters": [
{
"entity_type": "instance",
"operations": ["update", "delete"]
},
{
"entity_type": "pool"
},
],
"send_everything": false
}
This would return only update and delete events for instances and all events
for pools. Alternatively you can ask GARM to send you everything:
{
"send_everything": true
}
Signed-off-by: Gabriel Adrian Samfira <gsamfira@cloudbasesolutions.com>
148 lines
3.1 KiB
Go
148 lines
3.1 KiB
Go
// Copyright 2017 The Gorilla WebSocket Authors. All rights reserved.
|
|
// Use of this source code is governed by a BSD-style
|
|
// license that can be found in the LICENSE file.
|
|
|
|
package websocket
|
|
|
|
import (
|
|
"compress/flate"
|
|
"errors"
|
|
"io"
|
|
"strings"
|
|
"sync"
|
|
)
|
|
|
|
const (
|
|
minCompressionLevel = -2 // flate.HuffmanOnly not defined in Go < 1.6
|
|
maxCompressionLevel = flate.BestCompression
|
|
defaultCompressionLevel = 1
|
|
)
|
|
|
|
var (
|
|
flateWriterPools [maxCompressionLevel - minCompressionLevel + 1]sync.Pool
|
|
flateReaderPool = sync.Pool{New: func() interface{} {
|
|
return flate.NewReader(nil)
|
|
}}
|
|
)
|
|
|
|
func decompressNoContextTakeover(r io.Reader) io.ReadCloser {
|
|
const tail =
|
|
// Add four bytes as specified in RFC
|
|
"\x00\x00\xff\xff" +
|
|
// Add final block to squelch unexpected EOF error from flate reader.
|
|
"\x01\x00\x00\xff\xff"
|
|
|
|
fr, _ := flateReaderPool.Get().(io.ReadCloser)
|
|
fr.(flate.Resetter).Reset(io.MultiReader(r, strings.NewReader(tail)), nil)
|
|
return &flateReadWrapper{fr}
|
|
}
|
|
|
|
func isValidCompressionLevel(level int) bool {
|
|
return minCompressionLevel <= level && level <= maxCompressionLevel
|
|
}
|
|
|
|
func compressNoContextTakeover(w io.WriteCloser, level int) io.WriteCloser {
|
|
p := &flateWriterPools[level-minCompressionLevel]
|
|
tw := &truncWriter{w: w}
|
|
fw, _ := p.Get().(*flate.Writer)
|
|
if fw == nil {
|
|
fw, _ = flate.NewWriter(tw, level)
|
|
} else {
|
|
fw.Reset(tw)
|
|
}
|
|
return &flateWriteWrapper{fw: fw, tw: tw, p: p}
|
|
}
|
|
|
|
// truncWriter is an io.Writer that writes all but the last four bytes of the
|
|
// stream to another io.Writer.
|
|
type truncWriter struct {
|
|
w io.WriteCloser
|
|
n int
|
|
p [4]byte
|
|
}
|
|
|
|
func (w *truncWriter) Write(p []byte) (int, error) {
|
|
n := 0
|
|
|
|
// fill buffer first for simplicity.
|
|
if w.n < len(w.p) {
|
|
n = copy(w.p[w.n:], p)
|
|
p = p[n:]
|
|
w.n += n
|
|
if len(p) == 0 {
|
|
return n, nil
|
|
}
|
|
}
|
|
|
|
m := len(p)
|
|
if m > len(w.p) {
|
|
m = len(w.p)
|
|
}
|
|
|
|
if nn, err := w.w.Write(w.p[:m]); err != nil {
|
|
return n + nn, err
|
|
}
|
|
|
|
copy(w.p[:], w.p[m:])
|
|
copy(w.p[len(w.p)-m:], p[len(p)-m:])
|
|
nn, err := w.w.Write(p[:len(p)-m])
|
|
return n + nn, err
|
|
}
|
|
|
|
type flateWriteWrapper struct {
|
|
fw *flate.Writer
|
|
tw *truncWriter
|
|
p *sync.Pool
|
|
}
|
|
|
|
func (w *flateWriteWrapper) Write(p []byte) (int, error) {
|
|
if w.fw == nil {
|
|
return 0, errWriteClosed
|
|
}
|
|
return w.fw.Write(p)
|
|
}
|
|
|
|
func (w *flateWriteWrapper) Close() error {
|
|
if w.fw == nil {
|
|
return errWriteClosed
|
|
}
|
|
err1 := w.fw.Flush()
|
|
w.p.Put(w.fw)
|
|
w.fw = nil
|
|
if w.tw.p != [4]byte{0, 0, 0xff, 0xff} {
|
|
return errors.New("websocket: internal error, unexpected bytes at end of flate stream")
|
|
}
|
|
err2 := w.tw.w.Close()
|
|
if err1 != nil {
|
|
return err1
|
|
}
|
|
return err2
|
|
}
|
|
|
|
type flateReadWrapper struct {
|
|
fr io.ReadCloser
|
|
}
|
|
|
|
func (r *flateReadWrapper) Read(p []byte) (int, error) {
|
|
if r.fr == nil {
|
|
return 0, io.ErrClosedPipe
|
|
}
|
|
n, err := r.fr.Read(p)
|
|
if err == io.EOF {
|
|
// Preemptively place the reader back in the pool. This helps with
|
|
// scenarios where the application does not call NextReader() soon after
|
|
// this final read.
|
|
r.Close()
|
|
}
|
|
return n, err
|
|
}
|
|
|
|
func (r *flateReadWrapper) Close() error {
|
|
if r.fr == nil {
|
|
return io.ErrClosedPipe
|
|
}
|
|
err := r.fr.Close()
|
|
flateReaderPool.Put(r.fr)
|
|
r.fr = nil
|
|
return err
|
|
}
|