Function UpdateSubscriptions

pgserver/logrepl/subscription.go:29–104

(ctx *sql.Context)

var subscriptionMap = sync.Map{}

func UpdateSubscriptions(ctx *sql.Context) error {
	rows, err := adapter.QueryCatalog(ctx, catalog.InternalTables.PgSubscription.SelectAllStmt())
	if err != nil {
		return err
	}
	defer rows.Close()

	var subMap = make(map[string]*Subscription)
	for rows.Next() {
		var name, conn, pub, lsn string
		var enabled bool
		if err := rows.Scan(&name, &conn, &pub, &lsn, &enabled); err != nil {
			return err
		}
		subMap[name] = &Subscription{
			Subscription: name,
			Conn:         conn,
			Publication:  pub,
			LsnStr:       lsn,
			Enabled:      enabled,
			Replicator:   nil,
		}
	}

	if err = rows.Err(); err != nil {
		return err
	}

	for tempName, tempSub := range subMap {
		if _, loaded := subscriptionMap.LoadOrStore(tempName, tempSub); !loaded {
			replicator, err := NewLogicalReplicator(tempName, tempSub.Conn)
			if err != nil {
				return fmt.Errorf("failed to create logical replicator: %v", err)
			}

			if sub, ok := subscriptionMap.Load(tempName); ok {
				if subscription, ok := sub.(*Subscription); ok {
					subscription.Replicator = replicator
				}
			}

			err = replicator.CreateReplicationSlotIfNotExists(tempSub.Publication)
			if err != nil {
				return fmt.Errorf("failed to create replication slot: %v", err)
			}
			if tempSub.Enabled {
				go replicator.StartReplication(ctx, tempSub.Publication)
			}
		} else {
			if sub, ok := subscriptionMap.Load(tempName); ok {
				if subscription, ok := sub.(*Subscription); ok {
					if tempSub.Enabled != subscription.Enabled {
						subscription.Enabled = tempSub.Enabled
						if subscription.Enabled {
							go subscription.Replicator.StartReplication(ctx, subscription.Publication)
						} else {
							subscription.Replicator.Stop()
						}