strbuf

README.md

Strbuf

A small lib to help process and keep a small history of the past STDOUT/STDERR to show off, as well as a line parser to help ease the work needed to get functions called if lines appear in STDOUT/STDERR.

License

Copyright 2022 The Deadly Surgeon

Licensed under the Apache License, Version 2.0 (the "License"); you may not use these files except in compliance with the License. You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

buffer.go

package strbuf

import (
	"sync"
)

// Buffer holds onto strings and pops them when needed
type Buffer struct {
	lock      sync.RWMutex
	data      []string
	line      string
	stdinLine string

	*LineCallback
}

// Alloc changes the size of the buffer
func (b *Buffer) Alloc(size int) {
	b.data = make([]string, size)
}

// Add adds a new string to the buffer
func (b *Buffer) Add(s string) {
	b.lock.Lock()
	defer b.lock.Unlock()
	b.data = append(b.data[1:], s)
}

// Copy returns a copy of the array
// Might be worth iterating over data instead of copying it.
// The only issue doing that is if the socket can no longer write...
// Big dumb
func (b *Buffer) Copy() []string {
	b.lock.RLock()
	defer b.lock.RUnlock()

	tmp := make([]string, len(b.data))
	copy(tmp, b.data)
	return tmp
}

// Write attempts to emulate this bit of code except poorly
// https://golang.org/src/bufio/bufio.go?s=10157:10224#L377
//
// We probably should attempt to write this better.
func (b *Buffer) Write(d []byte) (n int, err error) {
	for i := 0; i < len(d); i++ {
		if d[i] != '\n' {
			b.line += string(d[i])
			continue
		}
		// Flush
		// TODO:
		// Handle straddlin \r
		// if len(b.line) > 0 && b.line[len(b.line)-1] == '\r' {
		// 	b.line = b.line[:len(b.line)-1]
		// }

		b.Add(b.line)
		b.Line(b.line)
		b.line = ""
	}

	return len(d), nil
}

func (b *Buffer) Read(d []byte) (n int, err error) {
	for i := 0; i < len(d); i++ {
		if d[i] != '\n' {
			b.stdinLine += string(d[i])
			continue
		}
		// Flush
		// TODO:
		// Handle straddlin \r
		// if len(b.stdinLine) > 0 && b.stdinLine[len(b.stdinLine)-1] == '\r' {
		// 	b.stdinLine = b.stdinLine[:len(b.line)-1]
		// }

		b.Add(b.stdinLine)
		b.Line(b.stdinLine)
		b.stdinLine = ""
	}

	return len(d), nil
}

// NewBuffer creates a new string buffer
func NewBuffer(size int) *Buffer {
	b := &Buffer{
		LineCallback: NewLineCallback(),
	}
	b.Alloc(size)

	return b
}

linecallback.go

package strbuf

import (
	"fmt"
	"sync"
)

// CallbackMatch stores the match condition
type CallbackMatch struct {
	matcher   func(string, string) bool
	condition string
	callback  func(string)
	async     bool
}

// Match returns if the string is a match or not
func (cbm *CallbackMatch) Match(line string) bool {
	return cbm.matcher(line, cbm.condition)
}

// Call executes the callback
func (cbm *CallbackMatch) Call(line string) {
	if cbm.async {
		go cbm.callback(line)
		return
	}
	cbm.callback(line)
}

// LineCallback calls back functions if it finds lines
type LineCallback struct {
	callbackLock *sync.RWMutex
	callbacks    []*CallbackMatch

	running bool
	lines   chan string
}

// Add adds the callback to the list.
// Doesn't protect against duplicates.
func (lcb *LineCallback) Add(
	matcher func(string, string) bool,
	condition string,
	callback func(string),
	async bool,
) {
	lcb.callbackLock.Lock()
	defer lcb.callbackLock.Unlock()

	lcb.callbacks = append(lcb.callbacks, &CallbackMatch{
		matcher:   matcher,
		condition: condition,
		callback:  callback,
		async:     async,
	})
}

// AddCallback adds the callback to the list.
// Doesn't protect against duplicates.
func (lcb *LineCallback) AddCallback(callback *CallbackMatch) {
	lcb.callbackLock.Lock()
	defer lcb.callbackLock.Unlock()

	lcb.callbacks = append(lcb.callbacks, callback)
}

// Line executes the LineCallback checks
func (lcb *LineCallback) Line(line string) {
	lcb.callbackLock.RLock()
	defer lcb.callbackLock.RUnlock()

	if lcb.lines != nil {
		if lcb.running {
			// can panic c:
			lcb.lines <- line
		}
		return
	}

	for i := 0; i < len(lcb.callbacks); i++ {
		if lcb.callbacks[i].Match(line) {
			lcb.callbacks[i].Call(line)
		}
	}
}

// service is a worker to process incoming lines
func (lcb *LineCallback) service() {
	for line := range lcb.lines {
		func() {
			lcb.callbackLock.RLock()
			defer lcb.callbackLock.RUnlock()

			for i := 0; i < len(lcb.callbacks); i++ {
				if lcb.callbacks[i].Match(line) {
					lcb.callbacks[i].Call(line)
				}
			}
		}()
	}
}

// Start starts the LineCallback as a worker to process linmes async
func (lcb *LineCallback) Start(buffersize int) error {
	lcb.callbackLock.Lock()
	defer lcb.callbackLock.Unlock()

	if lcb.running {
		return fmt.Errorf("LineCallback service has already running")
	}

	lcb.running = true
	lcb.lines = make(chan string, buffersize)

	go lcb.service()

	return nil
}

// Stop stops the LineCallback
func (lcb *LineCallback) Stop() error {
	lcb.callbackLock.Lock()
	defer lcb.callbackLock.Unlock()

	if !lcb.running {
		return fmt.Errorf("LineCallback service has already stopped")
	}

	// Can panic c:
	// Don't null this out, as we need it to determine if our object was set up
	// as a worker or not
	close(lcb.lines)

	return nil
}

// Close stops the LineCallback. Synonymous with LineCallback::Stop()
// This exists so that LineCallback can satasify an io.Closer
func (lcb *LineCallback) Close() error {
	return lcb.Stop()
}

// NewLineCallback creates a new LineCallback manager
func NewLineCallback() *LineCallback {
	return &LineCallback{
		callbacks:    []*CallbackMatch{},
		callbackLock: &sync.RWMutex{},
	}
}