layer1: add namespace dispatch step 17

pkg/utils/namespace_manager.go

@@ -22,6 +22,8 @@ type NamespaceManager interface {
 	Subscribe(subscriber NamespaceSubscriber)
 	SnapshotSubscribers() []string
 	Upsert(event NamespaceEvent) NamespaceRecord
+	DispatchAdd(ctx context.Context, namespace string) (NamespaceRecord, bool, error)
+	DispatchResync(ctx context.Context, namespace string) (NamespaceRecord, bool, error)
 	MarkPartState(namespace string, part string, state NamespacePartState) (NamespaceRecord, bool)
 	MarkPartRegistering(namespace string, part string) (NamespaceRecord, bool)
 	MarkPartActive(namespace string, part string) (NamespaceRecord, bool)
@@ -85,6 +87,23 @@ func (m *inMemoryNamespaceManager) SnapshotSubscribers() []string {
 	return names
 }
 
+func (m *inMemoryNamespaceManager) snapshotSubscriberObjects() []NamespaceSubscriber {
+	m.mu.RLock()
+	defer m.mu.RUnlock()
+
+	names := make([]string, 0, len(m.subs))
+	for name := range m.subs {
+		names = append(names, name)
+	}
+	sort.Strings(names)
+
+	subscribers := make([]NamespaceSubscriber, 0, len(names))
+	for _, name := range names {
+		subscribers = append(subscribers, m.subs[name])
+	}
+	return subscribers
+}
+
 func (m *inMemoryNamespaceManager) Snapshot() []string {
 	m.mu.RLock()
 	defer m.mu.RUnlock()
@@ -170,6 +189,43 @@ func (m *inMemoryNamespaceManager) Upsert(event NamespaceEvent) NamespaceRecord
 	return record.Clone()
 }
 
+func (m *inMemoryNamespaceManager) DispatchAdd(ctx context.Context, namespace string) (NamespaceRecord, bool, error) {
+	return m.dispatch(ctx, namespace, func(subscriber NamespaceSubscriber, record NamespaceRecord) error {
+		return subscriber.OnNamespaceAdd(ctx, record)
+	})
+}
+
+func (m *inMemoryNamespaceManager) DispatchResync(ctx context.Context, namespace string) (NamespaceRecord, bool, error) {
+	return m.dispatch(ctx, namespace, func(subscriber NamespaceSubscriber, record NamespaceRecord) error {
+		return subscriber.OnNamespaceResync(ctx, record)
+	})
+}
+
+func (m *inMemoryNamespaceManager) dispatch(ctx context.Context, namespace string, handler func(NamespaceSubscriber, NamespaceRecord) error) (NamespaceRecord, bool, error) {
+	record, ok := m.Get(namespace)
+	if !ok {
+		return NamespaceRecord{}, false, nil
+	}
+
+	var firstErr error
+	for _, subscriber := range m.snapshotSubscriberObjects() {
+		_, _ = m.MarkPartRegistering(namespace, subscriber.Name())
+		currentRecord, _ := m.Get(namespace)
+		err := handler(subscriber, currentRecord)
+		if err != nil {
+			_, _ = m.MarkPartFailed(namespace, subscriber.Name(), err)
+			if firstErr == nil {
+				firstErr = err
+			}
+			continue
+		}
+		_, _ = m.MarkPartActive(namespace, subscriber.Name())
+	}
+
+	record, _ = m.Get(namespace)
+	return record, true, firstErr
+}
+
 func (m *inMemoryNamespaceManager) MarkPartState(namespace string, part string, state NamespacePartState) (NamespaceRecord, bool) {
 	m.mu.Lock()
 	defer m.mu.Unlock()