writer.go 1.3 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869
  1. package buf
  2. import (
  3. "io"
  4. "sync"
  5. )
  6. // BufferToBytesWriter is a Writer that writes alloc.Buffer into underlying writer.
  7. type BufferToBytesWriter struct {
  8. writer io.Writer
  9. }
  10. // Write implements Writer.Write(). Write() takes ownership of the given buffer.
  11. func (v *BufferToBytesWriter) Write(buffer *Buffer) error {
  12. defer buffer.Release()
  13. for {
  14. nBytes, err := v.writer.Write(buffer.Bytes())
  15. if err != nil {
  16. return err
  17. }
  18. if nBytes == buffer.Len() {
  19. break
  20. }
  21. buffer.SliceFrom(nBytes)
  22. }
  23. return nil
  24. }
  25. // Release implements Releasable.Release().
  26. func (v *BufferToBytesWriter) Release() {
  27. v.writer = nil
  28. }
  29. type BytesToBufferWriter struct {
  30. sync.Mutex
  31. writer Writer
  32. }
  33. func (v *BytesToBufferWriter) Write(payload []byte) (int, error) {
  34. v.Lock()
  35. defer v.Unlock()
  36. if v.writer == nil {
  37. return 0, io.ErrClosedPipe
  38. }
  39. bytesWritten := 0
  40. size := len(payload)
  41. for size > 0 {
  42. buffer := New()
  43. nBytes, _ := buffer.Write(payload)
  44. size -= nBytes
  45. payload = payload[nBytes:]
  46. bytesWritten += nBytes
  47. err := v.writer.Write(buffer)
  48. if err != nil {
  49. return bytesWritten, err
  50. }
  51. }
  52. return bytesWritten, nil
  53. }
  54. // Release implements Releasable.Release()
  55. func (v *BytesToBufferWriter) Release() {
  56. v.Lock()
  57. v.writer.Release()
  58. v.writer = nil
  59. v.Unlock()
  60. }