fixup connected status

Signed-off-by: Vasiliy Tolstov <v.tolstov@unistack.org>

kgo.go

@@ -142,8 +142,9 @@ func (k *Broker) connect(ctx context.Context, opts ...kgo.Opt) (*kgo.Client, *ho
 			}
 			return nil, nil, err
 		}
-	}
+		k.connected.Store(1)
 		return c, htracer, nil
+	}
 }
 
 func (k *Broker) Connect(ctx context.Context) error {
@@ -267,7 +268,7 @@ func (k *Broker) publish(ctx context.Context, msgs []*broker.Message, opts ...br
 		k.connected.Store(1)
 	}
 	k.Unlock()
-	fmt.Printf("EEE\n")
+
 	options := broker.NewPublishOptions(opts...)
 	records := make([]*kgo.Record, 0, len(msgs))
 	var errs []string
@@ -322,9 +323,9 @@ func (k *Broker) publish(ctx context.Context, msgs []*broker.Message, opts ...br
 		return nil
 	}
 	ts := time.Now()
-	fmt.Printf("SSSSSSEEE\n")
+
 	results := k.c.ProduceSync(ctx, records...)
-	fmt.Printf("SSSSSS\n")
+
 	te := time.Since(ts)
 	for _, result := range results {
 		k.opts.Meter.Summary(semconv.PublishMessageLatencyMicroseconds, "endpoint", result.Record.Topic, "topic", result.Record.Topic).Update(te.Seconds())
@@ -404,6 +405,7 @@ func (k *Broker) Subscribe(ctx context.Context, topic string, handler broker.Han
 		consumers:    make(map[tp]*consumer),
 		done:         make(chan struct{}),
 		fatalOnError: fatalOnError,
+		connected:    k.connected,
 	}
 
 	kopts := append(k.kopts,

kgo_test.go

@@ -2,7 +2,6 @@ package kgo_test
 
 import (
 	"context"
-	"fmt"
 	"os"
 	"strings"
 	"sync/atomic"
@@ -214,7 +213,7 @@ func TestPubSub(t *testing.T) {
 				if prc := atomic.LoadInt64(&idx); prc == msgcnt {
 					close(done)
 				} else {
-					fmt.Printf("processed %v\n", prc)
+					t.Logf("processed %v\n", prc)
 				}
 			case <-ticker.C:
 				close(done)