| 12345678910111213141516171819202122232425262728293031323334353637383940414243444546 |
- package batch
- import (
- "io"
- "github.com/slackhq/nebula/util"
- )
- // Passthrough is a RxBatcher that doesn't batch anything, it just accumulates and then sends packets.
- type Passthrough struct {
- out io.Writer
- slots [][]byte
- arena *util.Arena
- cursor int
- }
- func NewPassthrough(w io.Writer, slots int, arena *util.Arena) *Passthrough {
- return &Passthrough{
- out: w,
- slots: make([][]byte, 0, slots),
- arena: arena,
- }
- }
- func (p *Passthrough) Reserve(sz int) []byte {
- return p.arena.Reserve(sz)
- }
- func (p *Passthrough) Commit(pkt []byte) error {
- p.slots = append(p.slots, pkt)
- return nil
- }
- func (p *Passthrough) Flush() error {
- var firstErr error
- for _, s := range p.slots {
- _, err := p.out.Write(s)
- if err != nil && firstErr == nil {
- firstErr = err
- }
- }
- clear(p.slots)
- p.slots = p.slots[:0]
- p.arena.Reset()
- return firstErr
- }
|