keybase/chat.go

package keybase

import (
	"bufio"
	"context"
	"fmt"
	"sync"

	"git.hugfreevikings.wtf/keybase/keybase/pkg/ctxreader"
)

type ChatAPI struct {
	sync.Mutex
	toStdin    chan []byte
	fromStdout chan msg
	fromListen chan []byte
	kbLoc      string
}

type msg interface{}

type apiMsg []byte

type errMsg struct {
	Err error
}

func (e errMsg) Error() string {
	return e.Err.Error()
}

func NewChatAPI(ctx context.Context, opts Options) (api *ChatAPI, err error) {
	if opts.ChannelBufferSize <= 0 {
		opts.ChannelBufferSize = 10
	}

	// set up the channels
	api = &ChatAPI{
		toStdin:    make(chan []byte, opts.ChannelBufferSize),
		fromStdout: make(chan msg, opts.ChannelBufferSize),
		fromListen: make(chan []byte, opts.ChannelBufferSize),
	}

	// get the basics
	if opts.KeybaseLocation == "" {
		opts.KeybaseLocation, err = locateKeybase()
		if err != nil {
			return nil, fmt.Errorf("failed to locate keybase: %v", err)
		}
	}
	api.kbLoc = opts.KeybaseLocation
	chatApiArgs, err := buildArgs(opts, "chat", "api")
	if err != nil {
		return nil, fmt.Errorf("failed to build api args: %v", err)
	}
	chatApiListenArgs, err := buildArgs(opts, "chat", "api-listen")
	if err != nil {
		return nil, fmt.Errorf("failed to build api-listen args: %v", err)
	}

	// build the commands
	chatApi, err := newApiCmd(ctx, api.kbLoc, chatApiArgs...)
	if err != nil {
		return nil, fmt.Errorf("failed to build api command")
	}
	chatListen, err := newApiCmd(ctx, api.kbLoc, chatApiListenArgs...)
	if err != nil {
		return nil, fmt.Errorf("failed to build api-listen command")
	}

	// create the goroutines
	// listen reader
	go func() {
		cr := ctxreader.NewContextReader(ctx, chatListen.Stdout())
		scanner := bufio.NewScanner(cr)
		for scanner.Scan() {
			api.fromListen <- scanner.Bytes()
		}
	}()

	// writing to stdin
	go func() {
		for {
			select {
			case msg := <-api.toStdin:
				_, err := chatApi.Stdin().Write(msg)
				if err != nil {
					api.fromStdout <- errMsg{Err: err}
				}
			case <-ctx.Done():
				return
			}
		}
	}()

	// reading from stdout
	go func() {
		cr := ctxreader.NewContextReader(ctx, chatApi.Stdout())
		scanner := bufio.NewScanner(cr)
		for scanner.Scan() {
			api.fromStdout <- scanner.Bytes()
		}
	}()

	// reading from stderr
	go func() {
		cr := ctxreader.NewContextReader(ctx, chatApi.Stderr())
		scanner := bufio.NewScanner(cr)
		for scanner.Scan() {
			api.fromStdout <- errMsg{Err: fmt.Errorf("%v", scanner.Text())}
		}
	}()

	// then start the cmds
	err = chatApi.Start()
	if err != nil {
		return nil, err
	}
	err = chatListen.Start()
	if err != nil {
		return nil, err
	}

	return
}

func (c *ChatAPI) SendRaw(ctx context.Context, msg []byte) ([]byte, error) {
	c.Lock()
	defer c.Unlock()

	// send message to api
	c.toStdin <- msg

	// wait for response
	for {
		select {
		case resp := <-c.fromStdout:
			switch resp := resp.(type) {
			case apiMsg:
				return resp, nil
			case errMsg:
				return nil, resp.Err
			}
		case <-ctx.Done():
			return nil, ctx.Err()
		}
	}
}