summaryrefslogtreecommitdiffstats
path: root/vendor/github.com/keybase
diff options
context:
space:
mode:
Diffstat (limited to 'vendor/github.com/keybase')
-rw-r--r--vendor/github.com/keybase/go-keybase-chat-bot/kbchat/chat.go12
-rw-r--r--vendor/github.com/keybase/go-keybase-chat-bot/kbchat/kbchat.go140
2 files changed, 97 insertions, 55 deletions
diff --git a/vendor/github.com/keybase/go-keybase-chat-bot/kbchat/chat.go b/vendor/github.com/keybase/go-keybase-chat-bot/kbchat/chat.go
index 129c43fa..bb2d26b8 100644
--- a/vendor/github.com/keybase/go-keybase-chat-bot/kbchat/chat.go
+++ b/vendor/github.com/keybase/go-keybase-chat-bot/kbchat/chat.go
@@ -122,7 +122,8 @@ func (a *API) GetTextMessages(channel chat1.ChatChannel, unreadOnly bool) ([]cha
return res, nil
}
-func (a *API) SendMessage(channel chat1.ChatChannel, body string, args ...interface{}) (SendResponse, error) {
+func (a *API) SendMessage(channel chat1.ChatChannel, body string, args ...interface{}) (resp SendResponse, err error) {
+ defer a.Trace(&err, "SendMessage")()
arg := newSendArg(sendMessageOptions{
Channel: channel,
Message: sendMessageBody{
@@ -139,7 +140,8 @@ func (a *API) Broadcast(body string, args ...interface{}) (SendResponse, error)
}, fmt.Sprintf(body, args...))
}
-func (a *API) SendMessageByConvID(convID chat1.ConvIDStr, body string, args ...interface{}) (SendResponse, error) {
+func (a *API) SendMessageByConvID(convID chat1.ConvIDStr, body string, args ...interface{}) (resp SendResponse, err error) {
+ defer a.Trace(&err, "SendMessageByConvID")()
arg := newSendArg(sendMessageOptions{
ConversationID: convID,
Message: sendMessageBody{
@@ -150,7 +152,8 @@ func (a *API) SendMessageByConvID(convID chat1.ConvIDStr, body string, args ...i
}
// SendMessageByTlfName sends a message on the given TLF name
-func (a *API) SendMessageByTlfName(tlfName string, body string, args ...interface{}) (SendResponse, error) {
+func (a *API) SendMessageByTlfName(tlfName string, body string, args ...interface{}) (resp SendResponse, err error) {
+ defer a.Trace(&err, "SendMessageByTlfName")()
arg := newSendArg(sendMessageOptions{
Channel: chat1.ChatChannel{
Name: tlfName,
@@ -162,7 +165,8 @@ func (a *API) SendMessageByTlfName(tlfName string, body string, args ...interfac
return a.doSend(arg)
}
-func (a *API) SendMessageByTeamName(teamName string, inChannel *string, body string, args ...interface{}) (SendResponse, error) {
+func (a *API) SendMessageByTeamName(teamName string, inChannel *string, body string, args ...interface{}) (resp SendResponse, err error) {
+ defer a.Trace(&err, "SendMessageByTeamName")()
channel := "general"
if inChannel != nil {
channel = *inChannel
diff --git a/vendor/github.com/keybase/go-keybase-chat-bot/kbchat/kbchat.go b/vendor/github.com/keybase/go-keybase-chat-bot/kbchat/kbchat.go
index b8d9eb3c..b5248106 100644
--- a/vendor/github.com/keybase/go-keybase-chat-bot/kbchat/kbchat.go
+++ b/vendor/github.com/keybase/go-keybase-chat-bot/kbchat/kbchat.go
@@ -6,6 +6,7 @@ import (
"errors"
"fmt"
"io"
+ "math"
"os"
"os/exec"
"runtime"
@@ -45,10 +46,10 @@ type Subscription struct {
}
func NewSubscription() *Subscription {
- newMsgsCh := make(chan SubscriptionMessage, 100)
- newConvsCh := make(chan SubscriptionConversation, 100)
- newWalletCh := make(chan SubscriptionWalletEvent, 100)
- errorCh := make(chan error, 100)
+ newMsgsCh := make(chan SubscriptionMessage, 250)
+ newConvsCh := make(chan SubscriptionConversation, 250)
+ newWalletCh := make(chan SubscriptionWalletEvent, 250)
+ errorCh := make(chan error, 250)
shutdownCh := make(chan struct{})
return &Subscription{
DebugOutput: NewDebugOutput("Subscription"),
@@ -137,6 +138,8 @@ type RunOptions struct {
EnableTyping bool
// Disable bot lite mode
DisableBotLiteMode bool
+ // Number of processes to spin up to connect to the keybase service
+ NumPipes int
}
func (r RunOptions) Location() string {
@@ -164,13 +167,20 @@ func Start(runOpts RunOptions, opts ...func(*API)) (*API, error) {
return api, nil
}
+type apiPipe struct {
+ sync.Mutex
+ input io.Writer
+ output *bufio.Reader
+ cmd *exec.Cmd
+}
+
// API is the main object used for communicating with the Keybase JSON API
type API struct {
sync.Mutex
*DebugOutput
- apiInput io.Writer
- apiOutput *bufio.Reader
- apiCmd *exec.Cmd
+ // Round robin hand out API pipes to allow concurrent API requests.
+ pipeIdx int
+ pipes []*apiPipe
username string
runOpts RunOptions
subscriptions []*Subscription
@@ -282,12 +292,15 @@ func (a *API) auth() (string, error) {
func (a *API) startPipes() (err error) {
a.Lock()
defer a.Unlock()
- if a.apiCmd != nil {
- if err := a.apiCmd.Process.Kill(); err != nil {
- return fmt.Errorf("unable to kill previous API command %v", err)
+ for _, pipe := range a.pipes {
+ if pipe.cmd != nil {
+ if err := pipe.cmd.Process.Kill(); err != nil {
+ return fmt.Errorf("unable to kill previous API command %v", err)
+ }
}
+ pipe.cmd = nil
}
- a.apiCmd = nil
+ a.pipes = nil
if a.runOpts.StartService {
args := []string{fmt.Sprintf("-enable-bot-lite-mode=%v", a.runOpts.DisableBotLiteMode), "service"}
@@ -306,30 +319,39 @@ func (a *API) startPipes() (err error) {
a.Debug("unable to set notifiation settings %v", err)
}
- a.apiCmd = a.runOpts.Command("chat", "api")
- if a.apiInput, err = a.apiCmd.StdinPipe(); err != nil {
- return fmt.Errorf("unable to get api stdin: %v", err)
- }
- output, err := a.apiCmd.StdoutPipe()
- if err != nil {
- return fmt.Errorf("unable to get api stdout: %v", err)
- }
- if runtime.GOOS != "windows" {
- a.apiCmd.ExtraFiles = []*os.File{output.(*os.File)}
- }
- if err := a.apiCmd.Start(); err != nil {
- return fmt.Errorf("unable to run chat api cmd: %v", err)
+ // Startup NumPipes processes to the keybase chat api
+ for i := 0; i < int(math.Max(float64(a.runOpts.NumPipes), 1)); i++ {
+ pipe := apiPipe{}
+ pipe.cmd = a.runOpts.Command("chat", "api")
+ if pipe.input, err = pipe.cmd.StdinPipe(); err != nil {
+ return fmt.Errorf("unable to get api stdin: %v", err)
+ }
+ output, err := pipe.cmd.StdoutPipe()
+ if err != nil {
+ return fmt.Errorf("unable to get api stdout: %v", err)
+ }
+ if runtime.GOOS != "windows" {
+ pipe.cmd.ExtraFiles = []*os.File{output.(*os.File)}
+ }
+ if err := pipe.cmd.Start(); err != nil {
+ return fmt.Errorf("unable to run chat api cmd: %v", err)
+ }
+ pipe.output = bufio.NewReader(output)
+ a.pipes = append(a.pipes, &pipe)
}
- a.apiOutput = bufio.NewReader(output)
return nil
}
-func (a *API) getAPIPipesLocked() (io.Writer, *bufio.Reader, error) {
- // this should only be called inside a lock
- if a.apiCmd == nil {
- return nil, nil, errAPIDisconnected
+func (a *API) getAPIPipes() (*apiPipe, error) {
+ a.Lock()
+ defer a.Unlock()
+ idx := a.pipeIdx % len(a.pipes)
+ a.pipeIdx++
+ pipe := a.pipes[idx]
+ if pipe.cmd == nil {
+ return nil, errAPIDisconnected
}
- return a.apiInput, a.apiOutput, nil
+ return pipe, nil
}
func (a *API) GetUsername() string {
@@ -337,21 +359,21 @@ func (a *API) GetUsername() string {
}
func (a *API) doSend(arg interface{}) (resp SendResponse, err error) {
- a.Lock()
- defer a.Unlock()
-
bArg, err := json.Marshal(arg)
if err != nil {
return SendResponse{}, fmt.Errorf("unable to send arg: %+v: %v", arg, err)
}
- input, output, err := a.getAPIPipesLocked()
+ pipe, err := a.getAPIPipes()
if err != nil {
return SendResponse{}, err
}
- if _, err := io.WriteString(input, string(bArg)); err != nil {
+ pipe.Lock()
+ defer pipe.Unlock()
+
+ if _, err := io.WriteString(pipe.input, string(bArg)); err != nil {
return SendResponse{}, err
}
- responseRaw, err := output.ReadBytes('\n')
+ responseRaw, err := pipe.output.ReadBytes('\n')
if err != nil {
return SendResponse{}, err
}
@@ -364,17 +386,17 @@ func (a *API) doSend(arg interface{}) (resp SendResponse, err error) {
}
func (a *API) doFetch(apiInput string) ([]byte, error) {
- a.Lock()
- defer a.Unlock()
-
- input, output, err := a.getAPIPipesLocked()
+ pipe, err := a.getAPIPipes()
if err != nil {
return nil, err
}
- if _, err := io.WriteString(input, apiInput); err != nil {
+ pipe.Lock()
+ defer pipe.Unlock()
+
+ if _, err := io.WriteString(pipe.input, apiInput); err != nil {
return nil, err
}
- byteOutput, err := output.ReadBytes('\n')
+ byteOutput, err := pipe.output.ReadBytes('\n')
if err != nil {
return nil, err
}
@@ -412,16 +434,22 @@ func (a *API) Listen(opts ListenOptions) (*Subscription, error) {
}
boutput.Scan()
t := boutput.Text()
+ submitErr := func(err error) {
+ if len(sub.errorCh)*2 > cap(sub.errorCh) {
+ a.Debug("large errorCh queue: len: %d cap: %d ", len(sub.errorCh), cap(sub.errorCh))
+ }
+ sub.errorCh <- err
+ }
var typeHolder TypeHolder
if err := json.Unmarshal([]byte(t), &typeHolder); err != nil {
- sub.errorCh <- fmt.Errorf("err: %v, data: %v", err, t)
+ submitErr(fmt.Errorf("err: %v, data: %v", err, t))
break
}
switch typeHolder.Type {
case "chat":
var notification chat1.MsgNotification
if err := json.Unmarshal([]byte(t), &notification); err != nil {
- sub.errorCh <- fmt.Errorf("err: %v, data: %v", err, t)
+ submitErr(fmt.Errorf("err: %v, data: %v", err, t))
break
}
if notification.Error != nil {
@@ -434,12 +462,15 @@ func (a *API) Listen(opts ListenOptions) (*Subscription, error) {
Channel: notification.Msg.Channel,
},
}
+ if len(sub.newMsgsCh)*2 > cap(sub.newMsgsCh) {
+ a.Debug("large newMsgsCh queue: len: %d cap: %d ", len(sub.newMsgsCh), cap(sub.newMsgsCh))
+ }
sub.newMsgsCh <- subscriptionMessage
}
case "chat_conv":
var notification chat1.ConvNotification
if err := json.Unmarshal([]byte(t), &notification); err != nil {
- sub.errorCh <- fmt.Errorf("err: %v, data: %v", err, t)
+ submitErr(fmt.Errorf("err: %v, data: %v", err, t))
break
}
if notification.Error != nil {
@@ -448,15 +479,21 @@ func (a *API) Listen(opts ListenOptions) (*Subscription, error) {
subscriptionConv := SubscriptionConversation{
Conversation: *notification.Conv,
}
+ if len(sub.newConvsCh)*2 > cap(sub.newConvsCh) {
+ a.Debug("large newConvsCh queue: len: %d cap: %d ", len(sub.newConvsCh), cap(sub.newConvsCh))
+ }
sub.newConvsCh <- subscriptionConv
}
case "wallet":
var holder PaymentHolder
if err := json.Unmarshal([]byte(t), &holder); err != nil {
- sub.errorCh <- fmt.Errorf("err: %v, data: %v", err, t)
+ submitErr(fmt.Errorf("err: %v, data: %v", err, t))
break
}
subscriptionPayment := SubscriptionWalletEvent(holder)
+ if len(sub.newWalletCh)*2 > cap(sub.newWalletCh) {
+ a.Debug("large newWalletCh queue: len: %d cap: %d ", len(sub.newWalletCh), cap(sub.newWalletCh))
+ }
sub.newWalletCh <- subscriptionPayment
default:
continue
@@ -518,7 +555,6 @@ func (a *API) Listen(opts ListenOptions) (*Subscription, error) {
}
boutput := bufio.NewScanner(output)
if err := p.Start(); err != nil {
-
a.Debug("Listen: failed to make listen scanner: %s", err)
time.Sleep(pause)
continue
@@ -568,10 +604,12 @@ func (a *API) Shutdown() (err error) {
for _, sub := range a.subscriptions {
sub.Shutdown()
}
- if a.apiCmd != nil {
- a.Debug("waiting for API command")
- if err := a.apiCmd.Wait(); err != nil {
- return err
+ for _, pipe := range a.pipes {
+ if pipe.cmd != nil {
+ a.Debug("waiting for API command")
+ if err := pipe.cmd.Wait(); err != nil {
+ return err
+ }
}
}
generated by cgit v1.2.3 (git 2.39.1) at 2024-12-02 17:04:57 +0000