// NewSource creates a new Source with the given random seed. Using the same
// seed should give the same series of events on a given version of Go.
775522
func NewSource(seed int64, concurrency int, max uint64) *Source {
	if max == 0 {
		max = math.MaxUint64
	}
	var n uint64
	s := &Source{
		events: make(chan *Event, 1000),
		closed: make(chan struct{}),
		n:      &n,
		max:    max,
	}
	for i := 0; i < concurrency; i++ {
		s.wg.Add(1)
		go func(i int) {
			defer s.wg.Done()
			// seed is multiplied by 10 because NewEventGenerator uses several
			// seeds internally (incrementing it by 1) and we'd prefer to avoid
			// re-using seeds between different generators.
			g := NewEventGenerator(seed * 10 * int64(i))
			for {
				select {
				case s.events <- g.Event():
				case <-s.closed:
					return
				}
			}
		}(i)
	}
	return s
}
372955
func (s *Service) UpdateSource(ctx context.Context, id influxdb.ID, upd influxdb.SourceUpdate) (*influxdb.Source, error) {
	var sr *influxdb.Source
	err := s.kv.Update(ctx, func(tx Tx) error {
		src, err := s.updateSource(ctx, tx, id, upd)
		if err != nil {
			return &influxdb.Error{
				Err: err,
			}
		}
		sr = src
		return nil
	})

	return sr, err
}
1001645
func (ls *lockedSource) Seed(seed int64) {
	ls.mu.Lock()
	ls.src.Seed(seed)
	ls.mu.Unlock()
}
1295915
func (s *PercolateService) Source(source string) *PercolateService {
	s.source = source
	return s
}
1208775
func Seed(seed int64) ListOption {
	return listOptionFunc(func(options *listOptions) {
		options.source = rand.NewSource(seed)
	})
}
457183
func (s *SearchService) SearchSource(searchSource *SearchSource) *SearchService {
	s.searchSource = searchSource
	if s.searchSource == nil {
		s.searchSource = NewSearchSource()
	}
	return s
}
1064904
func NewEventSource(w http.ResponseWriter) (*EventSource, error) {
	var err error
	es := &EventSource{
		closeNotify: make(chan bool),
		closed:      false,
	}
	if hijacker, ok := w.(http.Hijacker); ok {
		es.conn, _, err = hijacker.Hijack()
	} else {
		return nil, fmt.Errorf("responseWriter do not support hijacker")
	}

	if _, err = es.conn.Write([]byte("HTTP/1.1 200 OK\r\nContent-Type: text/event-stream\r\n")); err != nil {
		es.Close()
		return nil, err
	}

	if _, err = es.conn.Write([]byte("Vary: Accept-Encoding\r\n")); err != nil {
		es.Close()
		return nil, err
	}

	if _, err = es.conn.Write([]byte("\r\n")); err != nil {
		es.Close()
		return nil, err
	}

	go func() { // this goroutine to monitor connection whether closed
		io.Copy(nil, es.conn)
		es.Close()
	}()
	return es, nil
}
1149014
func (c *Client) UpdateSource(ctx context.Context, id platform.ID, upd platform.SourceUpdate) (*platform.Source, error) {
	var s *platform.Source
	err := c.db.Update(func(tx *bolt.Tx) error {
		src, err := c.updateSource(ctx, tx, id, upd)
		if err != nil {
			return &platform.Error{
				Err: err,
				Op:  getOp(platform.OpUpdateSource),
			}
		}
		s = src
		return nil
	})

	return s, err
}
148828
func (s *Service) SourceCreate(ctx context.Context) (*Source, error) {
	var source Source
	return &source, s.Post(ctx, &source, fmt.Sprintf("/sources"), nil)
}
591815
func (s *Staert) AddSource(src Source) {
	s.sources = append(s.sources, src)
}
1282584
func NewSourceInfo(bucket, object string, sse encrypt.ServerSide) SourceInfo {
	r := SourceInfo{
		bucket:     bucket,
		object:     object,
		start:      -1, // range is unspecified by default
		encryption: sse,
		Headers:    make(http.Header),
	}

	// Set the source header
	r.Headers.Set("x-amz-copy-source", s3utils.EncodePath(bucket+"/"+object))
	return r
}
109002
func NewSource(ctx *plan.Context, p *plan.Source) (*Source, error) {

	if p.Stmt == nil {
		return nil, fmt.Errorf("must have from for Source")
	}
	if p.Conn == nil {
		return nil, fmt.Errorf("Must have existing connection on Plan")
	}

	scanner, hasScanner := p.Conn.(schema.ConnScanner)

	// Some sources require context so we seed it here
	if sourceContext, needsContext := p.Conn.(RequiresContext); needsContext {
		sourceContext.SetContext(ctx)
	}

	if !hasScanner {
		e, hasSourceExec := p.Conn.(ExecutorSource)
		if hasSourceExec {
			s := &Source{
				TaskBase:   NewTaskBase(ctx),
				ExecSource: e,
				p:          p,
			}
			return s, nil
		}
		u.Warnf("source %T does not implement datasource.Scanner", p.Conn)
		return nil, fmt.Errorf("%T Must Implement Scanner for %q", p.Conn, p.Stmt.String())
	}
	s := &Source{
		TaskBase: NewTaskBase(ctx),
		Scanner:  scanner,
		p:        p,
	}
	return s, nil
}
1082285
func NewSearchSource() *SearchSource {
	return &SearchSource{
		from:         -1,
		size:         -1,
		aggregations: make(map[string]Aggregation),
		indexBoosts:  make(map[string]float64),
		innerHits:    make(map[string]*InnerHit),
	}
}
985411
func (s *Service) CreateSource(ctx context.Context, src *influxdb.Source) error {
	err := s.kv.Update(ctx, func(tx Tx) error {
		src.ID = s.IDGenerator.ID()

		// Generating an organization id if it missing or invalid
		if !src.OrganizationID.Valid() {
			src.OrganizationID = s.IDGenerator.ID()
		}

		return s.putSource(ctx, tx, src)
	})
	if err != nil {
		return &influxdb.Error{
			Err: err,
		}
	}
	return nil
}
1172971
func NewStorageSimpleStreamsDataSource(description string, storage StorageReader, basePath string, priority int, requireSigned bool) simplestreams.DataSource {
	return &storageSimpleStreamsDataSource{description, basePath, storage, false, priority, requireSigned}
}
28769
func NewSource(ctx *Context, stmt *rel.SqlSource, isFinal bool) (*Source, error) {
	s := &Source{Stmt: stmt, ctx: ctx, SourcePb: &SourcePb{Final: isFinal}, PlanBase: NewPlanBase(false)}
	err := s.load()
	if err != nil {
		return nil, err
	}
	return s, nil
}
808505
func NewSources(dir string, ids chronograf.ID, logger chronograf.Logger) chronograf.SourcesStore {
	return &Sources{
		Dir:     dir,
		Load:    load,
		Create:  create,
		ReadDir: ioutil.ReadDir,
		Remove:  os.Remove,
		IDs:     ids,
		Logger:  logger,
	}
}