// BuilderConnector satisfies the disco zk connect interface for a zkplus.Builder
162322
func BuilderConnector(b *zkplus.Builder) ZkConnCreator {
	return ZkConnCreatorFunc(func() (ZkConn, <-chan zk.Event, error) {
		return b.BuildDirect()
	})
}
1344960
func GetBuilder(c context.Context, bid BuilderID, limit int, cursor string) (*ui.Builder, error) {
	host, err := getHost(c)
	if err != nil {
		return nil, err
	}

	if limit < 0 {
		limit = 20
	}

	result := &ui.Builder{
		Name: bid.Builder,
	}
	if host == "debug" {
		return result, getDebugBuilds(c, bid, limit, result)
	}
	client, err := newBuildbucketClient(c, host)
	if err != nil {
		return nil, err
	}

	fetch := func(statusFilter string, limit int, cursor string) (result []*ui.BuildSummary, nextCursor string, err error) {
		msgs, nextCursor, err := fetchBuilds(c, client, bid, statusFilter, limit, cursor)
		if err != nil {
			logging.WithError(err).Errorf(c, "Could not fetch %s builds", statusFilter)
			return
		}
		result = toMiloBuildsSummaries(c, msgs)
		return
	}
	return result, parallel.FanOutIn(func(work chan<- func() error) {
		work <- func() error {
			return ensureDefined(c, host, bid)
		}
		work <- func() (err error) {
			result.MachinePool, err = getPool(c, bid)
			return
		}
		work <- func() (err error) {
			result.PendingBuilds, _, err = fetch(bbv1.StatusScheduled, -1, "")
			return
		}
		work <- func() (err error) {
			result.CurrentBuilds, _, err = fetch(bbv1.StatusStarted, -1, "")
			return
		}
		work <- func() (err error) {
			result.FinishedBuilds, result.NextCursor, err = fetch(bbv1.StatusCompleted, limit, cursor)
			result.PrevCursor = backCursor(c, bid, limit, cursor, result.NextCursor) // Safe to do even with error.
			return
		}
	})
}
491000
func (m *Migrator) boltConnect(file string) error {
	// Connect to the new BoltStore
	store, err := raftboltdb.NewBoltStore(file)
	if err != nil {
		return err
	}

	m.boltStore = store
	return nil
}
514148
func GetBuilder(c context.Context, masterName, builderName string, limit int, cursor string) (*ui.Builder, error) {
	if err := buildstore.CanAccessMaster(c, masterName); err != nil {
		return nil, err
	}
	result := &ui.Builder{
		Name: builderName,
	}
	master, err := buildstore.GetMaster(c, masterName, false)
	if err != nil {
		return nil, err
	}
	if clock.Now(c).Sub(master.Modified) > 2*time.Minute {
		warning := fmt.Sprintf(
			"WARNING: buildbotMasterEntry data is stale (last updated %s)", master.Modified)
		logging.Warningf(c, warning)
		result.Warning = warning
	}

	builder, ok := master.Builders[builderName]
	if !ok {
		// This long block is just to return a good error message when an invalid
		// buildbot builder is specified.
		keys := make([]string, 0, len(master.Builders))
		for k := range master.Builders {
			keys = append(keys, k)
		}
		sort.Strings(keys)
		// TODO(iannucci): add error-info-helper tags to give the error page enough
		// information to render link-to-master and link-to-builder.
		builders := strings.Join(keys, "\n")
		return nil, errors.Reason(
			"Cannot find builder %q in master %q.\nAvailable builders: \n%s",
			builderName, masterName, builders,
		).Tag(grpcutil.NotFoundTag).Err()
	}

	// Extract pending builds out of the master.
	result.PendingBuilds = make([]*ui.BuildSummary, len(builder.PendingBuildStates))
	result.PendingBuildNum = builder.PendingBuilds
	logging.Debugf(c, "Number of pending builds: %d", len(builder.PendingBuildStates))
	for i, pb := range builder.PendingBuildStates {
		start := time.Unix(int64(pb.SubmittedAt), 0).UTC()
		result.PendingBuilds[i] = &ui.BuildSummary{
			PendingTime: ui.NewInterval(c, start, time.Time{}),
			Blame:       make([]*ui.Commit, len(pb.Source.Changes)),
		}
		for j, cm := range pb.Source.Changes {
			result.PendingBuilds[i].Blame[j] = &ui.Commit{
				AuthorEmail: cm.Who,
				CommitURL:   cm.Revlink,
			}
		}
	}

	baseURL := "https://build.chromium.org/p/"
	if master.Internal {
		baseURL = "https://uberchromegw.corp.google.com/i/"
	}
	result.MachinePool = summarizeSlavePool(c, baseURL+master.Name, builder.Slaves, master.Slaves)

	return result, parallel.FanOutIn(func(work chan<- func() error) {
		q := buildstore.Query{
			Master:  masterName,
			Builder: builderName,

			NoAnnotationFetch: true,
		}
		work <- func() error {
			q := q
			q.Limit = limit
			q.Cursor = cursor
			q.Finished = buildstore.Yes
			res, err := buildstore.GetBuilds(c, q)
			if err != nil {
				return err
			}
			result.NextCursor = res.NextCursor
			result.PrevCursor = res.PrevCursor
			result.FinishedBuilds = make([]*ui.BuildSummary, len(res.Builds))
			for i, b := range res.Builds {
				result.FinishedBuilds[i] = renderBuild(c, b, false).BuildSummary()
			}
			return err
		}
		work <- func() error {
			q := q
			q.Finished = buildstore.No
			res, err := buildstore.GetBuilds(c, q)
			if err != nil {
				return err
			}
			result.CurrentBuilds = make([]*ui.BuildSummary, len(res.Builds))
			for i, b := range res.Builds {
				// currentBuilds is presented in reversed order, so flip it
				result.CurrentBuilds[len(res.Builds)-i-1] = renderBuild(c, b, false).BuildSummary()
			}
			return err
		}
	})
}
1201817
func (c *connectionTracker) AddConnection(conn net.Conn) {
	c.lock.Lock()
	defer c.lock.Unlock()
	c.conns[conn] = struct{}{}
}
682553
func newConnection() (*Conn, error) {
	c := new(Conn)
	if err := c.initConnection(); err != nil {
		return nil, err
	}

	return c, nil
}
1149801
func NewBuilder(w *Watcher, r *Runner) *Builder {
	return &Builder{watcher: w, runner: r}
}
445993
func newConnection(conn *spdystream.Connection, newStreamHandler httpstream.NewStreamHandler) httpstream.Connection {
	c := &connection{conn: conn, newStreamHandler: newStreamHandler}
	go conn.Serve(c.newSpdyStream)
	return c
}
1137902
func (p *Plugin) NewProtoConnection(name string) mux.Connection {
	return p.mux.NewProtoConnection(name, &keyval.SerializerJSON{})
}
1209719
func (c *Client) NewBuilder() *Builder {
	c.ensureTransmission()
	c.ensureBuilder()
	return c.builder.Clone()
}
442623
func NewDefaultConnector(config *ConnectorConfig) (*DefaultConnector, error) {
	if err := config.Validate(); err != nil {
		return nil, err
	}

	bootstrapConnections := make([]*BrokerConnection, len(config.BrokerList))
	for i := 0; i < len(config.BrokerList); i++ {
		broker := config.BrokerList[i]
		hostPort := strings.Split(broker, ":")
		if len(hostPort) != 2 {
			return nil, fmt.Errorf("incorrect broker connection string: %s", broker)
		}

		port, err := strconv.Atoi(hostPort[1])
		if err != nil {
			return nil, fmt.Errorf("incorrect port in broker connection string: %s", broker)
		}

		bootstrapConnections[i] = NewBrokerConnection(&Broker{
			ID:   -1,
			Host: hostPort[0],
			Port: int32(port),
		}, config.KeepAliveTimeout)
	}

	connector := &DefaultConnector{
		config:             *config,
		bootstrapBrokers:   bootstrapConnections,
		offsetCoordinators: make(map[string]int32),
	}
	connector.metadata = NewMetadata(connector, NewBrokers(config.KeepAliveTimeout), config.MetadataTTL)

	return connector, nil
}
20979
func FuzzConnAfterFromConfig(
	conn net.Conn,
	d time.Duration,
	config *config.FuzzConnConfig,
) net.Conn {
	return &FuzzedConnection{
		conn:   conn,
		start:  time.After(d),
		active: false,
		config: config,
	}
}
798941
func (r *Robot) AddConnection(c Connection) Connection {
	*r.connections = append(*r.Connections(), c)
	return c
}
623014
func ConnectConfig(addr string, config *nsq.Config) error {
	return DefaultProducer.ConnectConfig(addr, config)
}
739488
func (r *BinaryInstantiateREST) NewConnectOptions() (runtime.Object, bool, string) {
	return &buildapi.BinaryBuildRequestOptions{}, false, ""
}
128863
func SelectorConnectOption(s gosocks5.Selector) ConnectOption {
	return func(opts *ConnectOptions) {
		opts.Selector = s
	}
}
714523
func NewBuilder(client K8sClient, schema string) resolver.Builder {
	return &kubeBuilder{
		k8sClient: client,
		schema:    schema,
	}
}