chunk.go 3.0 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150
  1. package crypto
  2. import (
  3. "io"
  4. "v2ray.com/core/common"
  5. "v2ray.com/core/common/buf"
  6. "v2ray.com/core/common/serial"
  7. )
  8. type ChunkSizeDecoder interface {
  9. SizeBytes() int
  10. Decode([]byte) (uint16, error)
  11. }
  12. type ChunkSizeEncoder interface {
  13. SizeBytes() int
  14. Encode(uint16, []byte) []byte
  15. }
  16. type PlainChunkSizeParser struct{}
  17. func (PlainChunkSizeParser) SizeBytes() int {
  18. return 2
  19. }
  20. func (PlainChunkSizeParser) Encode(size uint16, b []byte) []byte {
  21. return serial.Uint16ToBytes(size, b)
  22. }
  23. func (PlainChunkSizeParser) Decode(b []byte) (uint16, error) {
  24. return serial.BytesToUint16(b), nil
  25. }
  26. type ChunkStreamReader struct {
  27. sizeDecoder ChunkSizeDecoder
  28. reader buf.Reader
  29. buffer []byte
  30. leftOver buf.MultiBuffer
  31. leftOverSize int
  32. }
  33. func NewChunkStreamReader(sizeDecoder ChunkSizeDecoder, reader io.Reader) *ChunkStreamReader {
  34. return &ChunkStreamReader{
  35. sizeDecoder: sizeDecoder,
  36. reader: buf.NewReader(reader),
  37. buffer: make([]byte, sizeDecoder.SizeBytes()),
  38. }
  39. }
  40. func (r *ChunkStreamReader) readAtLeast(size int) error {
  41. mb := r.leftOver
  42. r.leftOver = nil
  43. for mb.Len() < size {
  44. extra, err := r.reader.Read()
  45. if err != nil {
  46. mb.Release()
  47. return err
  48. }
  49. mb.AppendMulti(extra)
  50. }
  51. r.leftOver = mb
  52. return nil
  53. }
  54. func (r *ChunkStreamReader) readSize() (uint16, error) {
  55. if r.sizeDecoder.SizeBytes() > r.leftOver.Len() {
  56. if err := r.readAtLeast(r.sizeDecoder.SizeBytes() - r.leftOver.Len()); err != nil {
  57. return 0, err
  58. }
  59. }
  60. common.Must2(r.leftOver.Read(r.buffer))
  61. return r.sizeDecoder.Decode(r.buffer)
  62. }
  63. func (r *ChunkStreamReader) Read() (buf.MultiBuffer, error) {
  64. size := r.leftOverSize
  65. if size == 0 {
  66. nextSize, err := r.readSize()
  67. if err != nil {
  68. return nil, err
  69. }
  70. if nextSize == 0 {
  71. return nil, io.EOF
  72. }
  73. size = int(nextSize)
  74. }
  75. if r.leftOver.IsEmpty() {
  76. if err := r.readAtLeast(1); err != nil {
  77. return nil, err
  78. }
  79. }
  80. leftOverLen := r.leftOver.Len()
  81. if size >= leftOverLen {
  82. mb := r.leftOver
  83. r.leftOverSize = size - leftOverLen
  84. r.leftOver = nil
  85. return mb, nil
  86. }
  87. mb := r.leftOver.SliceBySize(size)
  88. mbLen := mb.Len()
  89. if mbLen != size {
  90. b := buf.New()
  91. common.Must(b.Reset(buf.ReadFullFrom(&r.leftOver, size-mbLen)))
  92. mb.Append(b)
  93. }
  94. r.leftOverSize = 0
  95. return mb, nil
  96. }
  97. type ChunkStreamWriter struct {
  98. sizeEncoder ChunkSizeEncoder
  99. writer buf.Writer
  100. }
  101. func NewChunkStreamWriter(sizeEncoder ChunkSizeEncoder, writer io.Writer) *ChunkStreamWriter {
  102. return &ChunkStreamWriter{
  103. sizeEncoder: sizeEncoder,
  104. writer: buf.NewWriter(writer),
  105. }
  106. }
  107. func (w *ChunkStreamWriter) Write(mb buf.MultiBuffer) error {
  108. mb2Write := buf.NewMultiBuffer()
  109. const sliceSize = 8192
  110. for {
  111. slice := mb.SliceBySize(sliceSize)
  112. b := buf.New()
  113. common.Must(b.AppendSupplier(func(buffer []byte) (int, error) {
  114. w.sizeEncoder.Encode(uint16(slice.Len()), buffer[:0])
  115. return w.sizeEncoder.SizeBytes(), nil
  116. }))
  117. mb2Write.Append(b)
  118. mb2Write.AppendMulti(slice)
  119. if mb.IsEmpty() {
  120. break
  121. }
  122. }
  123. return w.writer.Write(mb2Write)
  124. }