FilesExpand file tree

 master

/

unbuffered.go

Copy path

Blame

More file actions

Blame

More file actions

Latest commit

 

History

History

History

49 lines (44 loc) · 1.01 KB

 master

/

unbuffered.go

Copy path

Top

File metadata and controls

• Code
• Blame

49 lines (44 loc) · 1.01 KB

Raw

Copy raw file

Download raw file

Open symbols panel

Edit and raw actions

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

26

27

28

29

30

31

32

33

34

35

36

37

38

39

40

41

42

43

44

45

46

47

48

49

package broadcaster

import (

"io"

"sync"

)

// Unbuffered accumulates multiple io.WriteCloser by stream.

type Unbuffered struct {

mu sync.Mutex

writers []io.WriteCloser

}

// Add adds new io.WriteCloser.

func (w *Unbuffered) Add(writer io.WriteCloser) {

w.mu.Lock()

w.writers = append(w.writers, writer)

w.mu.Unlock()

}

// Write writes bytes to all writers. Failed writers will be evicted during

// this call.

func (w *Unbuffered) Write(p []byte) (n int, err error) {

w.mu.Lock()

var evict []int

for i, sw := range w.writers {

if n, err := sw.Write(p); err != nil || n != len(p) {

// On error, evict the writer

evict = append(evict, i)

}

}

for n, i := range evict {

w.writers = append(w.writers[:i-n], w.writers[i-n+1:]...)

}

w.mu.Unlock()

return len(p), nil

}

// Clean closes and removes all writers. Last non-eol-terminated part of data

// will be saved.

func (w *Unbuffered) Clean() error {

w.mu.Lock()

for _, sw := range w.writers {

sw.Close()

}

w.writers = nil

w.mu.Unlock()

return nil

}