<commit_before>// Copyright 2015 The Vanadium Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.

package discovery

import (
	"bytes"

	"v.io/v23"
	"v.io/v23/context"
	"v.io/v23/discovery"
	"v.io/v23/security"
)

// Scan implements discovery.Scanner.
func (ds *ds) Scan(ctx *context.T, query string) (<-chan discovery.Update, error) {
	// TODO(jhahn): Consider to use multiple target services so that the plugins
	// can filter advertisements more efficiently if possible.
	matcher, targetInterfaceName, err := newMatcher(ctx, query)
	if err != nil {
		return nil, err
	}

	ctx, cancel, err := ds.addTask(ctx)
	if err != nil {
		return nil, err
	}

	// TODO(jhahn): Revisit the buffer size.
	scanCh := make(chan Advertisement, 10)
	barrier := NewBarrier(func() {
		close(scanCh)
		ds.removeTask(ctx)
	})
	for _, plugin := range ds.plugins {
		if err := plugin.Scan(ctx, targetInterfaceName, scanCh, barrier.Add()); err != nil {
			cancel()
			return nil, err
		}
	}
	// TODO(jhahn): Revisit the buffer size.
	updateCh := make(chan discovery.Update, 10)
	go doScan(ctx, matcher, scanCh, updateCh)
	return updateCh, nil
}

func doScan(ctx *context.T, matcher matcher, scanCh <-chan Advertisement, updateCh chan<- discovery.Update) {
	defer close(updateCh)

	// Get the blessing names belong to the principal.
	//
	// TODO(jhahn): It isn't clear that we will always have the blessing required to decrypt
	// the advertisement as their "default" blessing - indeed it may not even be in the store.
	// Revisit this issue.
	principal := v23.GetPrincipal(ctx)
	var names []string
	if principal != nil {
		names = security.BlessingNames(principal, principal.BlessingStore().Default())
	}

	found := make(map[string]*Advertisement)
	for {
		select {
		case ad := <-scanCh:
			if err := decrypt(&ad, names); err != nil {
				// Couldn't decrypt it. Ignore it.
				if err != errNoPermission {
					ctx.Error(err)
				}
				continue
			}
			// Note that 'Lost' advertisement may not have full service information.
			// Thus we do not match the query against it. mergeAdvertisement() will
			// ignore it if it has not been scanned.
			if !ad.Lost {
				if !matcher.match(&ad) {
					continue
				}
			}
			for _, update := range mergeAdvertisement(found, &ad) {
				select {
				case updateCh <- update:
				case <-ctx.Done():
					return
				}
			}
		case <-ctx.Done():
			return
		}
	}
}

func mergeAdvertisement(found map[string]*Advertisement, ad *Advertisement) (updates []discovery.Update) {
	// The multiple plugins may return the same advertisements. We ignores the update
	// if it has been already sent through the update channel.
	prev := found[ad.Service.InstanceId]
	if ad.Lost {
		// TODO(jhahn): If some plugins return 'Lost' events for an advertisement update, we may
		// generates multiple 'Lost' and 'Found' events for the same update. In order to minimize
		// this flakiness, we may need to delay the handling of 'Lost'.
		if prev != nil {
			delete(found, ad.Service.InstanceId)
			updates = []discovery.Update{discovery.UpdateLost{discovery.Lost{Service: prev.Service}}}
		}
	} else {
		// TODO(jhahn): Need to compare the proximity as well.
		switch {
		case prev == nil:
			updates = []discovery.Update{discovery.UpdateFound{discovery.Found{Service: ad.Service}}}
		case !bytes.Equal(prev.Hash, ad.Hash):
			updates = []discovery.Update{
				discovery.UpdateLost{discovery.Lost{Service: prev.Service}},
				discovery.UpdateFound{discovery.Found{Service: copyService(&ad.Service)}},
			}
		}
		found[ad.Service.InstanceId] = ad
	}
	return
}
<commit_msg>discovery: fix a bug - adding a missed copyService()<commit_after>// Copyright 2015 The Vanadium Authors. All rights reserved.
// Use of this source code is governed by a BSD-style
// license that can be found in the LICENSE file.

package discovery

import (
	"bytes"

	"v.io/v23"
	"v.io/v23/context"
	"v.io/v23/discovery"
	"v.io/v23/security"
)

// Scan implements discovery.Scanner.
func (ds *ds) Scan(ctx *context.T, query string) (<-chan discovery.Update, error) {
	// TODO(jhahn): Consider to use multiple target services so that the plugins
	// can filter advertisements more efficiently if possible.
	matcher, targetInterfaceName, err := newMatcher(ctx, query)
	if err != nil {
		return nil, err
	}

	ctx, cancel, err := ds.addTask(ctx)
	if err != nil {
		return nil, err
	}

	// TODO(jhahn): Revisit the buffer size.
	scanCh := make(chan Advertisement, 10)
	barrier := NewBarrier(func() {
		close(scanCh)
		ds.removeTask(ctx)
	})
	for _, plugin := range ds.plugins {
		if err := plugin.Scan(ctx, targetInterfaceName, scanCh, barrier.Add()); err != nil {
			cancel()
			return nil, err
		}
	}
	// TODO(jhahn): Revisit the buffer size.
	updateCh := make(chan discovery.Update, 10)
	go doScan(ctx, matcher, scanCh, updateCh)
	return updateCh, nil
}

func doScan(ctx *context.T, matcher matcher, scanCh <-chan Advertisement, updateCh chan<- discovery.Update) {
	defer close(updateCh)

	// Get the blessing names belong to the principal.
	//
	// TODO(jhahn): It isn't clear that we will always have the blessing required to decrypt
	// the advertisement as their "default" blessing - indeed it may not even be in the store.
	// Revisit this issue.
	principal := v23.GetPrincipal(ctx)
	var names []string
	if principal != nil {
		names = security.BlessingNames(principal, principal.BlessingStore().Default())
	}

	found := make(map[string]*Advertisement)
	for {
		select {
		case ad := <-scanCh:
			if err := decrypt(&ad, names); err != nil {
				// Couldn't decrypt it. Ignore it.
				if err != errNoPermission {
					ctx.Error(err)
				}
				continue
			}
			// Note that 'Lost' advertisement may not have full service information.
			// Thus we do not match the query against it. mergeAdvertisement() will
			// ignore it if it has not been scanned.
			if !ad.Lost {
				if !matcher.match(&ad) {
					continue
				}
			}
			for _, update := range mergeAdvertisement(found, &ad) {
				select {
				case updateCh <- update:
				case <-ctx.Done():
					return
				}
			}
		case <-ctx.Done():
			return
		}
	}
}

func mergeAdvertisement(found map[string]*Advertisement, ad *Advertisement) (updates []discovery.Update) {
	// The multiple plugins may return the same advertisements. We ignores the update
	// if it has been already sent through the update channel.
	prev := found[ad.Service.InstanceId]
	if ad.Lost {
		// TODO(jhahn): If some plugins return 'Lost' events for an advertisement update, we may
		// generates multiple 'Lost' and 'Found' events for the same update. In order to minimize
		// this flakiness, we may need to delay the handling of 'Lost'.
		if prev != nil {
			delete(found, ad.Service.InstanceId)
			updates = []discovery.Update{discovery.UpdateLost{discovery.Lost{Service: prev.Service}}}
		}
	} else {
		// TODO(jhahn): Need to compare the proximity as well.
		service := copyService(&ad.Service)
		switch {
		case prev == nil:
			updates = []discovery.Update{discovery.UpdateFound{discovery.Found{Service: service}}}
		case !bytes.Equal(prev.Hash, ad.Hash):
			updates = []discovery.Update{
				discovery.UpdateLost{discovery.Lost{Service: prev.Service}},
				discovery.UpdateFound{discovery.Found{Service: service}},
			}
		}
		found[ad.Service.InstanceId] = ad
	}
	return
}

{'fraction_non_alphanumeric': 0.120837297811608, 'fraction_numerical': 0.0048932988990077475, 'mean_word_length': 3.4648058252427183, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1318479', 'n_tokens_mistral': 2663, 'n_tokens_neox': 2270, 'n_words': 763}