Method SubToVolume

src/bccore/pubsub.go:17–41

SubToVolume subscribes vol to q. This allows aribtrary volumes to be subscribed to arbitrary queues, but this is not the correct thing for a Node to do in the general case, Nodes should forward subscriptions to remote Volumes and Queues when both are on the same Node.

(ctx context.Context, qh blobcache.Handle, volh blobcache.Handle, spec blobcache.VolSubSpec)

// Nodes should forward subscriptions to remote Volumes and Queues when both
// are on the same Node.
func (sys *System) SubToVolume(ctx context.Context, qh blobcache.Handle, volh blobcache.Handle, spec blobcache.VolSubSpec) error {
	_, rights, err := sys.resolveVol(volh)
	if err != nil {
		return err
	}
	q, rights, err := sys.resolveQueue(qh, blobcache.Action_QUEUE_SUB_VOLUME)
	if err != nil {
		return err
	}
	sys.mu.Lock()
	defer sys.mu.Unlock()
	sub := sys.hub.Subscribe(volh.OID, func(ctx context.Context, k blobcache.OID, v *volume) {
		// TODO: use rights to tailor message
		_ = rights
		msg := blobcache.Message{}
		_, err := q.backend.Enqueue(ctx, []blobcache.Message{msg})
		if err != nil {
			logctx.Warn(ctx, "during subscription callback", zap.Error(err))
		}
	})
	// add the sub to the set of subs on the queue
	q.subs[sub] = struct{}{}
	// when q goes down, also need to unsubscribe the subs
	return nil
}

type hub struct {
	// mu guards pubs