writebuffer.go 2.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114
  1. package core
  2. import (
  3. "bytes"
  4. "io"
  5. "net/http"
  6. "sync"
  7. )
  8. // WriteBuffer by defaul Write(s) to bytes.Buffer.
  9. // But after WriteTo to new io.Writer - calls Reset.
  10. // Reset will flush current buffer data to new writer and starts to Write to new io.Writer
  11. // WriteTo will be locked until Write fails or Close will be called.
  12. type WriteBuffer struct {
  13. io.Writer
  14. err error
  15. mu sync.Mutex
  16. wg sync.WaitGroup
  17. state byte
  18. }
  19. func NewWriteBuffer(wr io.Writer) *WriteBuffer {
  20. if wr == nil {
  21. wr = bytes.NewBuffer(nil)
  22. }
  23. return &WriteBuffer{Writer: wr}
  24. }
  25. func (w *WriteBuffer) Write(p []byte) (n int, err error) {
  26. w.mu.Lock()
  27. if w.err != nil {
  28. err = w.err
  29. } else if n, err = w.Writer.Write(p); err != nil {
  30. w.err = err
  31. w.done()
  32. } else if f, ok := w.Writer.(http.Flusher); ok {
  33. f.Flush()
  34. }
  35. w.mu.Unlock()
  36. return
  37. }
  38. func (w *WriteBuffer) WriteTo(wr io.Writer) (n int64, err error) {
  39. w.Reset(wr)
  40. w.wg.Wait()
  41. return 0, w.err // TODO: fix counter
  42. }
  43. func (w *WriteBuffer) Close() error {
  44. if closer, ok := w.Writer.(io.Closer); ok {
  45. return closer.Close()
  46. }
  47. w.mu.Lock()
  48. w.done()
  49. w.mu.Unlock()
  50. return nil
  51. }
  52. func (w *WriteBuffer) Reset(wr io.Writer) {
  53. w.mu.Lock()
  54. w.add()
  55. if buf, ok := w.Writer.(*bytes.Buffer); ok && buf.Len() != 0 {
  56. if _, err := io.Copy(wr, buf); err != nil {
  57. w.err = err
  58. w.done()
  59. }
  60. }
  61. w.Writer = wr
  62. w.mu.Unlock()
  63. }
  64. const (
  65. none = iota
  66. start
  67. end
  68. )
  69. func (w *WriteBuffer) add() {
  70. if w.state == none {
  71. w.state = start
  72. w.wg.Add(1)
  73. }
  74. }
  75. func (w *WriteBuffer) done() {
  76. if w.state == start {
  77. w.state = end
  78. w.wg.Done()
  79. }
  80. }
  81. // OnceBuffer will catch only first message
  82. type OnceBuffer struct {
  83. buf []byte
  84. }
  85. func (o *OnceBuffer) Write(p []byte) (n int, err error) {
  86. if o.buf == nil {
  87. o.buf = p
  88. }
  89. return 0, io.EOF
  90. }
  91. func (o *OnceBuffer) WriteTo(w io.Writer) (n int64, err error) {
  92. return io.Copy(w, bytes.NewReader(o.buf))
  93. }
  94. func (o *OnceBuffer) Buffer() []byte {
  95. return o.buf
  96. }
  97. func (o *OnceBuffer) Len() int {
  98. return len(o.buf)
  99. }