jx/internal/containerlog/streamreader.go

// Copyright 2024 Matthew Rich <matthewrich.conf@gmail.com>. All rights reserved.

package containerlog

import (
	"bytes"
	"io"
)

type StreamReader struct {
	source io.Reader
	streams []*ReadBuffer
}

type ReadBuffer struct {
	streamtype StreamType
	reader *StreamReader
	bytes.Buffer
}

func NewStreamReader(source io.Reader) (s *StreamReader) {
	s = &StreamReader{
		source: source,
		streams: make([]*ReadBuffer, 3),
	}
	s.streams[StreamStdout] = &ReadBuffer{
		streamtype: StreamStdout,
		reader: s,
	}
	s.streams[StreamStderr] = &ReadBuffer{
		streamtype: StreamStderr,
		reader: s,
	}
	return
}

func (s *StreamReader) StdoutPipe() io.ReadCloser {
	return s.streams[StreamStdout]
}

func (s *StreamReader) StderrPipe() io.ReadCloser {
	return s.streams[StreamStderr]
}

func (b *ReadBuffer) Read(p []byte) (n int, e error) {
	for {
		if b.reader.streams[b.streamtype].Len() >= len(p) {
			break
		}

		streamtype, message, err := Read(b.reader.source)

		if err != nil {
			break
		}

		if b.reader.streams[streamtype] == nil {
			b.reader.streams[streamtype] = &ReadBuffer{
				streamtype: streamtype,
				reader: b.reader,
			}
		}

		if bytesRead, bufferErr := b.reader.streams[streamtype].WriteString(message); bytesRead != len(message) || bufferErr != nil {
			break
		}
	}

	return b.Buffer.Read(p)
}

func (b *ReadBuffer) Close() error {
	return nil
}