// NewStreamWithSelectConn opens a new Stream on a connection selected
// by selConn.
971146
func (s *Swarm) NewStreamSelectConn(selConn SelectConn) (*Stream, error) {
	if selConn == nil {
		return nil, errors.New("nil SelectConn")
	}

	conns := s.Conns()
	if len(conns) == 0 {
		return nil, ErrNoConnections
	}
	return s.newStreamSelectConn(selConn, conns)
}
600638
func WithReconnect(value bool) options.Opt {
	return func(p options.Params) {
		if setter, ok := p.(reconnectSetter); ok {
			setter.SetReconnect(value)
		}
	}
}
463293
func WithConn(conn net.PacketConn) ClientOpt {
	return func(c *Client) error {
		c.conn = conn
		return nil
	}
}
78303
func WithConnect(address string, onClose func()) Opt {
	return func(ctx context.Context, config shim.Config) (shimapi.ShimService, io.Closer, error) {
		conn, err := connect(address, annonDialer)
		if err != nil {
			return nil, nil, err
		}
		client := ttrpc.NewClient(conn, ttrpc.WithOnClose(onClose))
		return shimapi.NewShimClient(client), conn, nil
	}
}
426989
func (s *EventsService) SelectWithOffset(r RRSetKey, query string, offset int) ([]EventInfoDTO, ResultInfo, *http.Response, error) {
	var tld EventInfoListDTO

	uri := r.EventsQueryURI(query, offset)
	res, err := s.client.get(uri, &tld)

	pis := []EventInfoDTO{}
	for _, pi := range tld.Events {
		pis = append(pis, pi)
	}
	return pis, tld.Resultinfo, res, err
}
971066
func NewStream(ctx context.Context, request Request, opts ...CallOption) (Stream, error) {
	return DefaultClient.Stream(ctx, request, opts...)
}
322165
func NewStreamC(dataShards, parityShards int, conReads, conWrites bool, o ...Option) (StreamEncoder, error) {
	enc, err := New(dataShards, parityShards, o...)
	if err != nil {
		return nil, err
	}
	rs := enc.(*reedSolomon)
	r := rsStream{r: rs, bs: 4 << 20}
	r.readShards = readShards
	r.writeShards = writeShards
	if conReads {
		r.readShards = cReadShards
	}
	if conWrites {
		r.writeShards = cWriteShards
	}
	return &r, err
}
1171219
func ServeConnWithContext(conn io.ReadWriteCloser, context interface{}) {
	rpc.ServeCodecWithContext(NewServerCodec(conn), context)
}
128863
func SelectorConnectOption(s gosocks5.Selector) ConnectOption {
	return func(opts *ConnectOptions) {
		opts.Selector = s
	}
}
968177
func NewRocForStreamWithSrcLen(sourceLength uint, priceStream gotrade.DOHLCVStreamSubscriber, timePeriod int, selectData gotrade.DOHLCVDataSelectionFunc) (indicator *Roc, err error) {
	ind, err := NewRocWithSrcLen(sourceLength, timePeriod, selectData)
	priceStream.AddTickSubscription(ind)
	return ind, err
}
513228
func NewStdDevForStream(priceStream gotrade.DOHLCVStreamSubscriber, timePeriod int, selectData gotrade.DOHLCVDataSelectionFunc) (indicator *StdDev, err error) {
	ind, err := NewStdDev(timePeriod, selectData)
	priceStream.AddTickSubscription(ind)
	return ind, err
}
1204336
func NewServerConnection(conn net.Conn, newStreamHandler httpstream.NewStreamHandler) (httpstream.Connection, error) {
	spdyConn, err := spdystream.NewConnection(conn, true)
	if err != nil {
		defer conn.Close()
		return nil, err
	}

	return newConnection(spdyConn, newStreamHandler), nil
}
1064540
func (t *http2Client) NewStream(ctx context.Context, callHdr *CallHdr) (_ *Stream, err error) {
	ctx = peer.NewContext(ctx, t.getPeer())
	headerFields, err := t.createHeaderFields(ctx, callHdr)
	if err != nil {
		return nil, err
	}
	s := t.newStream(ctx, callHdr)
	cleanup := func(err error) {
		if s.swapState(streamDone) == streamDone {
			// If it was already done, return.
			return
		}
		// The stream was unprocessed by the server.
		atomic.StoreUint32(&s.unprocessed, 1)
		s.write(recvMsg{err: err})
		close(s.done)
		// If headerChan isn't closed, then close it.
		if atomic.SwapUint32(&s.headerDone, 1) == 0 {
			close(s.headerChan)
		}

	}
	hdr := &headerFrame{
		hf:        headerFields,
		endStream: false,
		initStream: func(id uint32) (bool, error) {
			t.mu.Lock()
			if state := t.state; state != reachable {
				t.mu.Unlock()
				// Do a quick cleanup.
				err := error(errStreamDrain)
				if state == closing {
					err = ErrConnClosing
				}
				cleanup(err)
				return false, err
			}
			t.activeStreams[id] = s
			if channelz.IsOn() {
				atomic.AddInt64(&t.czData.streamsStarted, 1)
				atomic.StoreInt64(&t.czData.lastStreamCreatedTime, time.Now().UnixNano())
			}
			var sendPing bool
			// If the number of active streams change from 0 to 1, then check if keepalive
			// has gone dormant. If so, wake it up.
			if len(t.activeStreams) == 1 && t.keepaliveEnabled {
				select {
				case t.awakenKeepalive <- struct{}{}:
					sendPing = true
					// Fill the awakenKeepalive channel again as this channel must be
					// kept non-writable except at the point that the keepalive()
					// goroutine is waiting either to be awaken or shutdown.
					t.awakenKeepalive <- struct{}{}
				default:
				}
			}
			t.mu.Unlock()
			return sendPing, nil
		},
		onOrphaned: cleanup,
		wq:         s.wq,
	}
	firstTry := true
	var ch chan struct{}
	checkForStreamQuota := func(it interface{}) bool {
		if t.streamQuota <= 0 { // Can go negative if server decreases it.
			if firstTry {
				t.waitingStreams++
			}
			ch = t.streamsQuotaAvailable
			return false
		}
		if !firstTry {
			t.waitingStreams--
		}
		t.streamQuota--
		h := it.(*headerFrame)
		h.streamID = t.nextID
		t.nextID += 2
		s.id = h.streamID
		s.fc = &inFlow{limit: uint32(t.initialWindowSize)}
		if t.streamQuota > 0 && t.waitingStreams > 0 {
			select {
			case t.streamsQuotaAvailable <- struct{}{}:
			default:
			}
		}
		return true
	}
	var hdrListSizeErr error
	checkForHeaderListSize := func(it interface{}) bool {
		if t.maxSendHeaderListSize == nil {
			return true
		}
		hdrFrame := it.(*headerFrame)
		var sz int64
		for _, f := range hdrFrame.hf {
			if sz += int64(f.Size()); sz > int64(*t.maxSendHeaderListSize) {
				hdrListSizeErr = status.Errorf(codes.Internal, "header list size to send violates the maximum size (%d bytes) set by server", *t.maxSendHeaderListSize)
				return false
			}
		}
		return true
	}
	for {
		success, err := t.controlBuf.executeAndPut(func(it interface{}) bool {
			if !checkForStreamQuota(it) {
				return false
			}
			if !checkForHeaderListSize(it) {
				return false
			}
			return true
		}, hdr)
		if err != nil {
			return nil, err
		}
		if success {
			break
		}
		if hdrListSizeErr != nil {
			return nil, hdrListSizeErr
		}
		firstTry = false
		select {
		case <-ch:
		case <-s.ctx.Done():
			return nil, ContextErr(s.ctx.Err())
		case <-t.goAway:
			return nil, errStreamDrain
		case <-t.ctx.Done():
			return nil, ErrConnClosing
		}
	}
	if t.statsHandler != nil {
		outHeader := &stats.OutHeader{
			Client:      true,
			FullMethod:  callHdr.Method,
			RemoteAddr:  t.remoteAddr,
			LocalAddr:   t.localAddr,
			Compression: callHdr.SendCompress,
		}
		t.statsHandler.HandleRPC(s.ctx, outHeader)
	}
	return s, nil
}
1183674
func (sam *SAM) NewStreamSession(id string, keys I2PKeys, options []string) (*StreamSession, error) {
	conn, err := sam.newGenericSession("STREAM", id, keys, options, []string{})
	if err != nil {
		return nil, err
	}
	return &StreamSession{sam.address, id, conn, keys}, nil
}
1216142
func WithConnectionEvent(value chan *dispatcher.ConnectionEvent) options.Opt {
	return func(p options.Params) {
		if setter, ok := p.(connectEventChSetter); ok {
			setter.SetConnectEventCh(value)
		}
	}
}
1339567
func (l *Listener) rejectConn(err error) {
	select {
	case l.conns <- maybeConn{err: err}:
	case <-l.quit:
	}
}
542244
func (qre *QueryExecutor) execSelect() (*sqltypes.Result, error) {
	if qre.plan.Fields != nil {
		result, err := qre.qFetch(qre.logStats, qre.plan.FullQuery, qre.bindVars)
		if err != nil {
			return nil, err
		}
		// result is read-only. So, let's copy it before modifying.
		newResult := *result
		newResult.Fields = qre.plan.Fields
		return &newResult, nil
	}
	conn, err := qre.getConn()
	if err != nil {
		return nil, err
	}
	defer conn.Recycle()
	return qre.dbConnFetch(conn, qre.plan.FullQuery, qre.bindVars, "", true)
}