passthrough.go 861 B

12345678910111213141516171819202122232425262728293031323334353637383940414243444546
  1. package batch
  2. import (
  3. "io"
  4. "github.com/slackhq/nebula/util"
  5. )
  6. // Passthrough is a RxBatcher that doesn't batch anything, it just accumulates and then sends packets.
  7. type Passthrough struct {
  8. out io.Writer
  9. slots [][]byte
  10. arena *util.Arena
  11. cursor int
  12. }
  13. func NewPassthrough(w io.Writer, slots int, arena *util.Arena) *Passthrough {
  14. return &Passthrough{
  15. out: w,
  16. slots: make([][]byte, 0, slots),
  17. arena: arena,
  18. }
  19. }
  20. func (p *Passthrough) Reserve(sz int) []byte {
  21. return p.arena.Reserve(sz)
  22. }
  23. func (p *Passthrough) Commit(pkt []byte) error {
  24. p.slots = append(p.slots, pkt)
  25. return nil
  26. }
  27. func (p *Passthrough) Flush() error {
  28. var firstErr error
  29. for _, s := range p.slots {
  30. _, err := p.out.Write(s)
  31. if err != nil && firstErr == nil {
  32. firstErr = err
  33. }
  34. }
  35. clear(p.slots)
  36. p.slots = p.slots[:0]
  37. p.arena.Reset()
  38. return firstErr
  39. }