// startRelation creates a new data value for tracking details of the
// relation and starts watching the related models for subnets added or removed.
325731
func (fw *Firewaller) startRelation(rel *params.RemoteRelation, role charm.RelationRole) error {
	remoteApps, err := fw.remoteRelationsApi.RemoteApplications([]string{rel.RemoteApplicationName})
	if err != nil {
		return errors.Trace(err)
	}
	remoteAppResult := remoteApps[0]
	if remoteAppResult.Error != nil {
		return errors.Trace(err)
	}

	tag := names.NewRelationTag(rel.Key)
	data := &remoteRelationData{
		fw:                  fw,
		tag:                 tag,
		remoteModelUUID:     rel.SourceModelUUID,
		localApplicationTag: names.NewApplicationTag(rel.ApplicationName),
		endpointRole:        role,
		relationReady:       make(chan remoteRelationInfo),
	}

	// Start the worker which will watch the remote relation for things like new networks.
	if err := fw.relationWorkerRunner.StartWorker(tag.Id(), func() (worker.Worker, error) {
		if err := catacomb.Invoke(catacomb.Plan{
			Site: &data.catacomb,
			Work: data.watchLoop,
		}); err != nil {
			return nil, errors.Trace(err)
		}
		return data, nil
	}); err != nil {
		return errors.Annotate(err, "error starting remote relation worker")
	}
	fw.relationIngress[tag] = data

	data.isOffer = remoteAppResult.Result.IsConsumerProxy
	return fw.startRelationPoller(rel.Key, rel.RemoteApplicationName, data.relationReady)
}
788793
func (r *Relation) WatchRelationIngressNetworks() StringsWatcher {
	return newrelationNetworksWatcher(r.st, r.Tag().Id(), ingress)
}
929583
func (u *UniterAPI) goalStateRelations(appName, principalName string, allRelations []*state.Relation) (map[string]params.UnitsGoalState, error) {

	result := map[string]params.UnitsGoalState{}

	for _, r := range allRelations {
		statusInfo, err := r.Status()
		if err != nil {
			return nil, errors.Annotate(err, "getting relation status")
		}
		endPoints := r.Endpoints()
		if len(endPoints) == 1 {
			// Ignore peer relations here.
			continue
		}

		// First determine the local endpoint name to use later
		// as the key in the result map.
		var resultEndpointName string
		for _, e := range endPoints {
			if e.ApplicationName == appName {
				resultEndpointName = e.Name
			}
		}
		if resultEndpointName == "" {
			continue
		}

		// Now gather the goal state.
		for _, e := range endPoints {
			var key string
			app, err := u.st.Application(e.ApplicationName)
			if err == nil {
				key = app.Name()
			} else if errors.IsNotFound(err) {
				logger.Debugf("application %q must be a remote application.", e.ApplicationName)
				remoteApplication, err := u.st.RemoteApplication(e.ApplicationName)
				if err != nil {
					return nil, err
				}
				var ok bool
				key, ok = remoteApplication.URL()
				if !ok {
					// If we are on the offering side of a remote relation, don't show anything
					// in goal state for that relation.
					continue
				}
			} else {
				return nil, err
			}

			// We don't show units for the same application as we are currently processing.
			if key == appName {
				continue
			}

			goalState := params.GoalStateStatus{}
			goalState.Status = statusInfo.Status.String()
			goalState.Since = statusInfo.Since
			relationGoalState := result[e.Name]
			if relationGoalState == nil {
				relationGoalState = params.UnitsGoalState{}
			}
			relationGoalState[key] = goalState

			// For local applications, add in the status of units as well.
			if app != nil {
				units, err := u.goalStateUnits(app, principalName)
				if err != nil {
					return nil, err
				}
				for unitName, unitGS := range units {
					relationGoalState[unitName] = unitGS
				}
			}

			// Merge in the goal state for the current remote endpoint
			// with any other goal state already collected for the local endpoint.
			unitsGoalState := result[resultEndpointName]
			if unitsGoalState == nil {
				unitsGoalState = params.UnitsGoalState{}
			}
			for k, v := range relationGoalState {
				unitsGoalState[k] = v
			}
			result[resultEndpointName] = unitsGoalState
		}
	}
	return result, nil
}
1230994
func WatchEgressAddressesForRelations(resources facade.Resources, st State, relations params.Entities) (params.StringsWatchResults, error) {
	results := params.StringsWatchResults{
		make([]params.StringsWatchResult, len(relations.Entities)),
	}

	one := func(tag string) (id string, changes []string, _ error) {
		logger.Debugf("Watching egress addresses for %+v", tag)

		relationTag, err := names.ParseRelationTag(tag)
		if err != nil {
			return "", nil, errors.Trace(err)
		}

		// Load the relation details for the current token.
		localEndpoint, err := localApplication(st, relationTag)
		if err != nil {
			return "", nil, errors.Trace(err)
		}

		w, err := NewEgressAddressWatcher(st, localEndpoint.relation, localEndpoint.application)
		if err != nil {
			return "", nil, errors.Trace(err)
		}

		// TODO(wallyworld) - we will need to watch subnets too, but only
		// when we support using cloud local addresses
		//filter := func(id interface{}) bool {
		//	include, err := includeAsIngressSubnet(id.(string))
		//	if err != nil {
		//		logger.Warningf("invalid CIDR %q", id)
		//	}
		//	return include
		//}
		//w := api.st.WatchSubnets(filter)

		changes, ok := <-w.Changes()
		if !ok {
			return "", nil, common.ServerError(watcher.EnsureErr(w))
		}
		return resources.Register(w), changes, nil
	}

	for i, e := range relations.Entities {
		watcherId, changes, err := one(e.Tag)
		if err != nil {
			results.Results[i].Error = common.ServerError(err)
			continue
		}
		results.Results[i].StringsWatcherId = watcherId
		results.Results[i].Changes = changes
	}
	return results, nil
}
1027647
func (w *relationUnitsWorker) Wait() error {
	err := w.catacomb.Wait()
	if err != nil {
		logger.Errorf("error in relation units worker for %v: %v", w.relationTag.Id(), err)
	}
	return err
}
820332
func (s *Invitation) SetRelationshipStatus(v string) *Invitation {
	s.RelationshipStatus = &v
	return s
}
997604
func (bot *TgBot) StartChain() *TgBot {
	bot.ChainConditionals = append(bot.ChainConditionals, NewChainStructure())
	bot.BuildingChain = true
	return bot
}
60008
func (api *CrossModelRelationsAPI) RelationUnitSettings(relationUnits params.RemoteRelationUnits) (params.SettingsResults, error) {
	results := params.SettingsResults{
		Results: make([]params.SettingsResult, len(relationUnits.RelationUnits)),
	}
	for i, arg := range relationUnits.RelationUnits {
		relationTag, err := api.st.GetRemoteEntity(arg.RelationToken)
		if err != nil {
			results.Results[i].Error = common.ServerError(err)
			continue
		}
		if err := api.checkMacaroonsForRelation(relationTag, arg.Macaroons); err != nil {
			results.Results[i].Error = common.ServerError(err)
			continue
		}

		ru := params.RelationUnit{
			Relation: relationTag.String(),
			Unit:     arg.Unit,
		}
		settings, err := commoncrossmodel.RelationUnitSettings(api.st, ru)
		if err != nil {
			results.Results[i].Error = common.ServerError(err)
			continue
		}
		results.Results[i].Settings = settings
	}
	return results, nil
}
379825
func (api *APIBase) CharmRelations(p params.ApplicationCharmRelations) (params.ApplicationCharmRelationsResults, error) {
	var results params.ApplicationCharmRelationsResults
	if err := api.checkCanRead(); err != nil {
		return results, errors.Trace(err)
	}

	app, err := api.backend.Application(p.ApplicationName)
	if err != nil {
		return results, errors.Trace(err)
	}
	endpoints, err := app.Endpoints()
	if err != nil {
		return results, errors.Trace(err)
	}
	results.CharmRelations = make([]string, len(endpoints))
	for i, endpoint := range endpoints {
		results.CharmRelations[i] = endpoint.Relation.Name
	}
	return results, nil
}
934317
func (s *ConfigurationItem) SetRelationships(v []*Relationship) *ConfigurationItem {
	s.Relationships = v
	return s
}
96449
func (rd *remoteRelationData) watchLoop() error {
	defer func() {
		if rd.crossModelFirewallerFacade != nil {
			rd.crossModelFirewallerFacade.Close()
		}
	}()

	// First, wait for relation to become ready.
	for rd.relationToken == "" {
		select {
		case <-rd.catacomb.Dying():
			return rd.catacomb.ErrDying()
		case remoteRelationInfo := <-rd.relationReady:
			rd.relationToken = remoteRelationInfo.relationToken
			rd.applicationToken = remoteRelationInfo.applicationToken
			logger.Debugf(
				"relation %v for remote app %v in model %v is ready",
				rd.relationToken, rd.applicationToken, rd.remoteModelUUID)
		}
	}

	if rd.endpointRole == charm.RoleRequirer {
		return rd.requirerEndpointLoop()
	}
	return rd.providerEndpointLoop()
}
23334
func (c *Client) MacaroonForRelation(relationKey string) (*macaroon.Macaroon, error) {
	relationTag := names.NewRelationTag(relationKey)
	args := params.Entities{[]params.Entity{{Tag: relationTag.String()}}}
	var results params.MacaroonResults
	err := c.facade.FacadeCall("MacaroonForRelations", args, &results)
	if err != nil {
		return nil, errors.Trace(err)
	}
	if len(results.Results) != 1 {
		return nil, errors.Errorf("expected 1 result, got %d", len(results.Results))
	}
	result := results.Results[0]
	if result.Error != nil {
		return nil, result.Error
	}
	return result.Result, nil
}
79408
func newRelationLifeSuspendedWatcher(
	backend modelBackend,
	members bson.D,
	filter func(key interface{}) bool,
	transform func(id string) string,
) *relationLifeSuspendedWatcher {
	w := &relationLifeSuspendedWatcher{
		commonWatcher: newCommonWatcher(backend),
		out:           make(chan []string),
		members:       members,
		filter:        filter,
		transform:     transform,
		lifeSuspended: make(map[string]relationLifeSuspended),
	}
	w.tomb.Go(func() error {
		defer close(w.out)
		return w.loop()
	})
	return w
}
116789
func (u *UniterAPIV6) JoinedRelations(args params.Entities) (params.StringsResults, error) {
	result := params.StringsResults{
		Results: make([]params.StringsResult, len(args.Entities)),
	}
	if len(args.Entities) == 0 {
		return result, nil
	}
	canRead, err := u.accessUnit()
	if err != nil {
		return params.StringsResults{}, err
	}
	for i, entity := range args.Entities {
		tag, err := names.ParseUnitTag(entity.Tag)
		if err != nil {
			result.Results[i].Error = common.ServerError(common.ErrPerm)
			continue
		}
		err = common.ErrPerm
		if canRead(tag) {
			var unit *state.Unit
			unit, err = u.getUnit(tag)
			if err == nil {
				result.Results[i].Result, err = relationsInScopeTags(unit)
			}
		}
		result.Results[i].Error = common.ServerError(err)
	}
	return result, nil
}
706242
func (st *State) WatchRelationUnits(
	relationTag names.RelationTag,
	unitTag names.UnitTag,
) (watcher.RelationUnitsWatcher, error) {
	var results params.RelationUnitsWatchResults
	args := params.RelationUnits{
		RelationUnits: []params.RelationUnit{{
			Relation: relationTag.String(),
			Unit:     unitTag.String(),
		}},
	}
	err := st.facade.FacadeCall("WatchRelationUnits", args, &results)
	if err != nil {
		return nil, err
	}
	if len(results.Results) != 1 {
		return nil, fmt.Errorf("expected 1 result, got %d", len(results.Results))
	}
	result := results.Results[0]
	if result.Error != nil {
		return nil, result.Error
	}
	w := apiwatcher.NewRelationUnitsWatcher(st.facade.RawAPICaller(), result)
	return w, nil
}
435027
func (n *natsConsumer) Start(acc telegraf.Accumulator) error {
	n.acc = acc.WithTracking(n.MaxUndeliveredMessages)

	var connectErr error

	// set default NATS connection options
	opts := nats.DefaultOptions

	// override max reconnection tries
	opts.MaxReconnect = -1

	// override servers if any were specified
	opts.Servers = n.Servers

	opts.Secure = n.Secure

	if n.conn == nil || n.conn.IsClosed() {
		n.conn, connectErr = opts.Connect()
		if connectErr != nil {
			return connectErr
		}

		// Setup message and error channels
		n.errs = make(chan error)
		n.conn.SetErrorHandler(n.natsErrHandler)

		n.in = make(chan *nats.Msg, 1000)
		for _, subj := range n.Subjects {
			sub, err := n.conn.QueueSubscribe(subj, n.QueueGroup, func(m *nats.Msg) {
				n.in <- m
			})
			if err != nil {
				return err
			}
			// ensure that the subscription has been processed by the server
			if err = n.conn.Flush(); err != nil {
				return err
			}
			// set the subscription pending limits
			if err = sub.SetPendingLimits(n.PendingMessageLimit, n.PendingBytesLimit); err != nil {
				return err
			}
			n.subs = append(n.subs, sub)
		}
	}

	ctx, cancel := context.WithCancel(context.Background())
	n.cancel = cancel

	// Start the message reader
	n.wg.Add(1)
	go func() {
		defer n.wg.Done()
		go n.receiver(ctx)
	}()

	log.Printf("I! Started the NATS consumer service, nats: %v, subjects: %v, queue: %v\n",
		n.conn.ConnectedUrl(), n.Subjects, n.QueueGroup)

	return nil
}
1142600
func NewRelationStatusWatcher(
	caller base.APICaller, result params.RelationLifeSuspendedStatusWatchResult,
) watcher.RelationStatusWatcher {
	w := &relationStatusWatcher{
		caller:                  caller,
		relationStatusWatcherId: result.RelationStatusWatcherId,
		out:                     make(chan []watcher.RelationStatusChange),
	}
	w.tomb.Go(func() error {
		return w.loop(result.Changes)
	})
	return w
}