micro/go-micro · error

not connected

Error message

not connected

What it means

The in-memory broker's Publish method refuses to publish when the broker has not been connected (or has been disconnected). memoryBroker tracks connectivity with a `connected` flag guarded by an RWMutex; Publish checks it first and short-circuits with this error. The in-memory broker still requires an explicit Connect() call before it will accept messages, unlike a naive expectation that it is always ready.

Source

Thrown at broker/memory.go:95

	}

	m.connected = false

	return nil
}

func (m *memoryBroker) Init(opts ...Option) error {
	for _, o := range opts {
		o(m.opts)
	}
	return nil
}

func (m *memoryBroker) Publish(topic string, msg *Message, opts ...PublishOption) error {
	m.RLock()
	if !m.connected {
		m.RUnlock()
		return errors.New("not connected")
	}

	subs, ok := m.Subscribers[topic]
	m.RUnlock()
	if !ok {
		return nil
	}

	var v interface{}
	if m.opts.Codec != nil {
		buf, err := m.opts.Codec.Marshal(msg)
		if err != nil {
			return err
		}
		v = buf
	} else {
		v = msg
	}

View on GitHub (pinned to 24529f1404)

Solutions

1. Call broker.Connect(context) before the first Publish and check its error.
2. If the broker may have been disconnected, call Connect() again (it is idempotent for the memory broker).
3. Ensure no code path calls Disconnect() while publishing is still expected; coordinate shutdown ordering.
4. Wrap Publish in a retry that calls Connect() on 'not connected' and re-publishes.

Example fix

// before
broker := memory.NewBroker()
_ = broker.Publish(context, "events", msg)

// after
broker := memory.NewBroker()
if err := broker.Connect(context); err != nil {
    log.Fatal(err)
}
if err := broker.Publish(context, "events", msg); err != nil {
    log.Fatal(err)
}

type connectable interface {
    Connect(context.Context) error
    Connected() bool // if exposed; otherwise track locally
}
func ensureConnected(b broker.Broker, ctx context.Context) error {
    if c, ok := b.(connectable); ok {
        return c.Connect(ctx)
    }
    return nil
}

func isConnected(err error) bool {
    return err != nil && strings.Contains(err.Error(), "not connected")
}

if err := br.Publish(ctx, topic, msg); err != nil {
    if strings.Contains(err.Error(), "not connected") {
        if cerr := br.Connect(ctx); cerr == nil {
            err = br.Publish(ctx, topic, msg)
        } else {
            err = cerr
        }
    }
    return err
}

When it happens

Trigger: Calling broker.Publish(topic, msg) before calling broker.Connect(), or after calling Disconnect(). Any Publish issued on a `memoryBroker` whose `m.connected` is false returns exactly this error from broker/memory.go:95.

Common situations: Scripts that construct a memory broker and immediately publish without Connect(); tests that share a broker across subtests where one subtest called Disconnect(); publishing from a goroutine that starts before broker initialization completes.

Related errors

• agent: checkpointed run is terminal with status
• cannot init while connected
• watcher stopped
• agent: StreamAsk unsupported by implementation
• agent: ResumeStreamAsk unsupported by implementation

AI-assisted analysis of micro/go-micro@24529f1404 (2026-09-01). Data as JSON: /api/errors/0f38048073667582. Report an issue: GitHub.