diff --git a/pool.go b/pool.go index f0d3e3f..b9435b8 100644 --- a/pool.go +++ b/pool.go @@ -256,13 +256,13 @@ func (pool *SimplePool) subManyEose(ctx context.Context, urls []string, filters case reason := <-sub.ClosedReason: if strings.HasPrefix(reason, "auth-required:") && pool.authHandler != nil && !hasAuthed { // relay is requesting auth. if we can we will perform auth and try again - if err := relay.Auth(ctx, pool.authHandler); err == nil { + err := relay.Auth(ctx, pool.authHandler) + if err == nil { hasAuthed = true // so we don't keep doing AUTH again and again goto subscribe } - } else { - log.Printf("CLOSED from %s: '%s'\n", nm, reason) } + log.Printf("CLOSED from %s: '%s'\n", nm, reason) return case evt, more := <-sub.Events: if !more { diff --git a/relay.go b/relay.go index fd3f7eb..f65f423 100644 --- a/relay.go +++ b/relay.go @@ -319,7 +319,9 @@ func (r *Relay) publish(ctx context.Context, id string, env Envelope) error { } // listen for an OK callback + gotOk := false r.okCallbacks.Store(id, func(ok bool, reason string) { + gotOk = true if !ok { err = fmt.Errorf("msg: %s", reason) } @@ -338,6 +340,9 @@ func (r *Relay) publish(ctx context.Context, id string, env Envelope) error { select { case <-ctx.Done(): // this will be called when we get an OK or when the context has been canceled + if gotOk { + return err + } return ctx.Err() case <-r.connectionContext.Done(): // this is caused when we lose connectivity