passthrough.go 1.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657
  1. package batch
  2. import (
  3. "io"
  4. "github.com/slackhq/nebula/udp"
  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. backing []byte
  11. cursor int
  12. }
  13. func NewPassthrough(w io.Writer) *Passthrough {
  14. const baseNumSlots = 128
  15. return &Passthrough{
  16. out: w,
  17. slots: make([][]byte, 0, baseNumSlots),
  18. backing: make([]byte, 0, baseNumSlots*udp.MTU),
  19. }
  20. }
  21. func (p *Passthrough) Reserve(sz int) []byte {
  22. if len(p.backing)+sz > cap(p.backing) {
  23. // Grow: allocate a fresh backing. Already-committed slices still
  24. // reference the old array and remain valid until Flush drops them.
  25. newCap := max(cap(p.backing)*2, sz)
  26. p.backing = make([]byte, 0, newCap)
  27. }
  28. start := len(p.backing)
  29. p.backing = p.backing[:start+sz]
  30. return p.backing[start : start+sz : start+sz] //return zero length, sz-cap slice
  31. }
  32. func (p *Passthrough) Commit(pkt []byte) error {
  33. p.slots = append(p.slots, pkt)
  34. return nil
  35. }
  36. func (p *Passthrough) Flush() error {
  37. var firstErr error
  38. for _, s := range p.slots {
  39. _, err := p.out.Write(s)
  40. if err != nil && firstErr == nil {
  41. firstErr = err
  42. }
  43. }
  44. for i := range p.slots {
  45. p.slots[i] = nil
  46. }
  47. p.slots = p.slots[:0]
  48. p.backing = p.backing[:0]
  49. return firstErr
  50. }