// StartMultipleStreamMixService 开启多人连麦服务
// {@link https://help.aliyun.com/document_detail/51313.html?spm=5176.doc52720.6.682.0eSfxw}
960653
func (l *Live) StartMultipleStreamMixService(streamName, mixTemplate string, resp interface{}) (err error) {
	req := l.cloneRequest(StartMultipleStreamMixServiceAction)
	req.SetArgs("StreamName", streamName)
	req.SetArgs("MixTemplate", mixTemplate)
	err = l.rpc.Query(req, resp)
	return
}
581228
func Start() {
	mux := &yarpcHTTPMux{
		handlers: make(map[string]http.Handler),
	}
	mux.HandleFunc("handlertimeout/raw", handlerTimeoutRawHandler)

	server = net.NewHTTPServer(
		&http.Server{
			Addr:         addr,
			Handler:      mux,
			ReadTimeout:  5 * time.Second,
			WriteTimeout: 5 * time.Second,
		})

	if err := server.ListenAndServe(); err != nil {
		log.Fatalf("failed to start HTTP server: %v", err)
	}
}
767305
func NewDefaultPlusDiForStreamWithSrcLen(sourceLength uint, priceStream gotrade.DOHLCVStreamSubscriber) (indicator *PlusDi, err error) {
	ind, err := NewDefaultPlusDiWithSrcLen(sourceLength)
	priceStream.AddTickSubscription(ind)
	return ind, err
}
1031218
func NewMultistream(c io.ReadWriteCloser, proto string) Multistream {
	return &lazyClientConn{
		protos: []string{proto},
		con:    c,
	}
}
1034184
func (conn *BytesConnectionStr) ConsumeTopic(msgClb func(message *client.ConsumerMessage), topics ...string) error {
	conn.multiplexer.rwlock.Lock()
	defer conn.multiplexer.rwlock.Unlock()

	if conn.multiplexer.started {
		return fmt.Errorf("ConsumeTopic can be called only if the multiplexer has not been started yet")
	}

	for _, topic := range topics {
		// check if we have already consumed the topic
		var found bool
		var subs *consumerSubscription
	LoopSubs:
		for _, subscription := range conn.multiplexer.mapping {
			if subscription.manual == true {
				// do not mix dynamic and manual mode
				continue
			}
			if subscription.topic == topic {
				found = true
				subs = subscription
				break LoopSubs
			}
		}

		if !found {
			subs = &consumerSubscription{
				manual:         false, // non-manual example
				topic:          topic,
				connectionName: conn.name,
				byteConsMsg:    msgClb,
			}
			// subscribe new topic
			conn.multiplexer.mapping = append(conn.multiplexer.mapping, subs)
		}

		// add subscription to consumerList
		subs.byteConsMsg = msgClb
	}

	return nil
}
484547
func RegisterStreamServiceHandler(ctx context.Context, mux *runtime.ServeMux, conn *grpc.ClientConn) error {
	return RegisterStreamServiceHandlerClient(ctx, mux, NewStreamServiceClient(conn))
}
534632
func (c *Catalog) ServiceMultipleTags(service string, tags []string, q *QueryOptions) ([]*CatalogService, *QueryMeta, error) {
	return c.service(service, tags, q, false)
}
316549
func (a *azureObjects) CompleteMultipartUpload(ctx context.Context, bucket, object, uploadID string, uploadedParts []minio.CompletePart, opts minio.ObjectOptions) (objInfo minio.ObjectInfo, err error) {
	metadataObject := getAzureMetadataObjectName(object, uploadID)
	if err = a.checkUploadIDExists(ctx, bucket, object, uploadID); err != nil {
		return objInfo, err
	}

	if err = checkAzureUploadID(ctx, uploadID); err != nil {
		return objInfo, err
	}

	var metadataReader io.Reader
	blob := a.client.GetContainerReference(bucket).GetBlobReference(metadataObject)
	if metadataReader, err = blob.Get(nil); err != nil {
		return objInfo, azureToObjectError(err, bucket, metadataObject)
	}

	var metadata azureMultipartMetadata
	if err = json.NewDecoder(metadataReader).Decode(&metadata); err != nil {
		logger.LogIf(ctx, err)
		return objInfo, azureToObjectError(err, bucket, metadataObject)
	}

	objBlob := a.client.GetContainerReference(bucket).GetBlobReference(object)

	var allBlocks []storage.Block
	for i, part := range uploadedParts {
		var partMetadataReader io.Reader
		var partMetadata partMetadataV1
		partMetadataObject := getAzureMetadataPartName(object, uploadID, part.PartNumber)
		pblob := a.client.GetContainerReference(bucket).GetBlobReference(partMetadataObject)
		if partMetadataReader, err = pblob.Get(nil); err != nil {
			return objInfo, azureToObjectError(err, bucket, partMetadataObject)
		}

		if err = json.NewDecoder(partMetadataReader).Decode(&partMetadata); err != nil {
			logger.LogIf(ctx, err)
			return objInfo, azureToObjectError(err, bucket, partMetadataObject)
		}

		if partMetadata.ETag != part.ETag {
			return objInfo, minio.InvalidPart{}
		}
		for _, blockID := range partMetadata.BlockIDs {
			allBlocks = append(allBlocks, storage.Block{ID: blockID, Status: storage.BlockStatusUncommitted})
		}
		if i < (len(uploadedParts)-1) && partMetadata.Size < azureS3MinPartSize {
			return objInfo, minio.PartTooSmall{
				PartNumber: uploadedParts[i].PartNumber,
				PartSize:   partMetadata.Size,
				PartETag:   uploadedParts[i].ETag,
			}
		}
	}

	err = objBlob.PutBlockList(allBlocks, nil)
	if err != nil {
		return objInfo, azureToObjectError(err, bucket, object)
	}
	objBlob.Metadata, objBlob.Properties, err = s3MetaToAzureProperties(ctx, metadata.Metadata)
	if err != nil {
		return objInfo, azureToObjectError(err, bucket, object)
	}
	objBlob.Metadata["md5sum"], err = cmd.ComputeCompleteMultipartMD5(uploadedParts)
	if err != nil {
		return objInfo, err
	}
	err = objBlob.SetProperties(nil)
	if err != nil {
		return objInfo, azureToObjectError(err, bucket, object)
	}
	err = objBlob.SetMetadata(nil)
	if err != nil {
		return objInfo, azureToObjectError(err, bucket, object)
	}
	var partNumberMarker int
	for {
		lpi, err := a.ListObjectParts(ctx, bucket, object, uploadID, partNumberMarker, maxPartsCount, minio.ObjectOptions{})
		if err != nil {
			break
		}
		for _, part := range lpi.Parts {
			pblob := a.client.GetContainerReference(bucket).GetBlobReference(
				getAzureMetadataPartName(object, uploadID, part.PartNumber))
			pblob.Delete(nil)
		}
		partNumberMarker = lpi.NextPartNumberMarker
		if !lpi.IsTruncated {
			break
		}
	}

	blob = a.client.GetContainerReference(bucket).GetBlobReference(metadataObject)
	derr := blob.Delete(nil)
	logger.GetReqInfo(ctx).AppendTags("uploadID", uploadID)
	logger.LogIf(ctx, derr)

	return a.GetObjectInfo(ctx, bucket, object, minio.ObjectOptions{})
}
578360
func (s *HTTPServer) fsStreamImpl(resp http.ResponseWriter,
	req *http.Request, method string, args interface{}, allocID string) (interface{}, error) {

	// Get the correct handler
	localClient, remoteClient, localServer := s.rpcHandlerForAlloc(allocID)
	var handler structs.StreamingRpcHandler
	var handlerErr error
	if localClient {
		handler, handlerErr = s.agent.Client().StreamingRpcHandler(method)
	} else if remoteClient {
		handler, handlerErr = s.agent.Client().RemoteStreamingRpcHandler(method)
	} else if localServer {
		handler, handlerErr = s.agent.Server().StreamingRpcHandler(method)
	}

	if handlerErr != nil {
		return nil, CodedError(500, handlerErr.Error())
	}

	// Create a pipe connecting the (possibly remote) handler to the http response
	httpPipe, handlerPipe := net.Pipe()
	decoder := codec.NewDecoder(httpPipe, structs.MsgpackHandle)
	encoder := codec.NewEncoder(httpPipe, structs.MsgpackHandle)

	// Create a goroutine that closes the pipe if the connection closes.
	ctx, cancel := context.WithCancel(req.Context())
	go func() {
		<-ctx.Done()
		httpPipe.Close()
	}()

	// Create an output that gets flushed on every write
	output := ioutils.NewWriteFlusher(resp)

	// Create a channel that decodes the results
	errCh := make(chan HTTPCodedError)
	go func() {
		defer cancel()

		// Send the request
		if err := encoder.Encode(args); err != nil {
			errCh <- CodedError(500, err.Error())
			return
		}

		for {
			select {
			case <-ctx.Done():
				errCh <- nil
				return
			default:
			}

			var res cstructs.StreamErrWrapper
			if err := decoder.Decode(&res); err != nil {
				errCh <- CodedError(500, err.Error())
				return
			}
			decoder.Reset(httpPipe)

			if err := res.Error; err != nil {
				if err.Code != nil {
					errCh <- CodedError(int(*err.Code), err.Error())
					return
				}
			}

			if _, err := io.Copy(output, bytes.NewReader(res.Payload)); err != nil {
				errCh <- CodedError(500, err.Error())
				return
			}
		}
	}()

	handler(handlerPipe)
	cancel()
	codedErr := <-errCh

	// Ignore EOF and ErrClosedPipe errors.
	if codedErr != nil &&
		(codedErr == io.EOF ||
			strings.Contains(codedErr.Error(), "closed") ||
			strings.Contains(codedErr.Error(), "EOF")) {
		codedErr = nil
	}
	return nil, codedErr
}
697509
func (conn *BytesManualConnectionStr) StartPostInitConsumer(topic string, partition int32, offset int64) (*sarama.PartitionConsumer, error) {
	multiplexer := conn.multiplexer
	multiplexer.WithFields(logging.Fields{"topic": topic}).Debugf("Post-init consuming started")

	if multiplexer.Consumer == nil || multiplexer.Consumer.SConsumer == nil {
		multiplexer.Warn("Unable to start post-init Consumer, client not available in the mux")
		return nil, nil
	}

	// Consumer that reads topic/partition/offset. Throws error if offset is 'in the future' (message with offset does not exist yet)
	partitionConsumer, err := multiplexer.Consumer.SConsumer.ConsumePartition(topic, partition, offset)
	if err != nil {
		return nil, err
	}
	multiplexer.Consumer.StartConsumerManualHandlers(partitionConsumer)

	return &partitionConsumer, nil
}
143358
func newMultipart(url string, method string, body interface{}, files ...File) (*http.Request, error) {

	bb := &bytes.Buffer{}
	writer := multipart.NewWriter(bb)
	defer writer.Close()
	for _, f := range files {
		part, err := writer.CreateFormFile(f.ParamName, f.FileName)
		if err != nil {
			return nil, errors.WithStack(err)
		}
		_, err = io.Copy(part, f)
	}

	for k, v := range toURLValues(body) {
		for _, vv := range v {
			err := writer.WriteField(k, vv)
			if err != nil {
				return nil, errors.WithStack(err)
			}
		}
	}

	req, err := http.NewRequest(method, url, bb)
	if err != nil {
		return nil, errors.WithStack(err)
	}
	req.Header.Set("Content-Type", writer.FormDataContentType())
	return req, nil
}
1146785
func NewMultilineAdapter(route *router.Route) (a router.LogAdapter, err error) {
	enableByDefault := true
	enableStr := os.Getenv("MULTILINE_ENABLE_DEFAULT")
	if enableStr != "" {
		var err error
		enableByDefault, err = strconv.ParseBool(enableStr)
		if err != nil {
			return nil, errors.New("multiline: invalid value for MULTILINE_ENABLE_DEFAULT (must be true|false): " + enableStr)
		}
	}

	pattern := os.Getenv("MULTILINE_PATTERN")
	if pattern == "" {
		pattern = `^\s`
	}

	separator := os.Getenv("MULTILINE_SEPARATOR")
	if separator == "" {
		separator = "\n"
	}
	patternRegexp, err := regexp.Compile(pattern)
	if err != nil {
		return nil, errors.New("multiline: invalid value for MULTILINE_PATTERN (must be regexp): " + pattern)
	}

	matchType := os.Getenv("MULTILINE_MATCH")
	if matchType == "" {
		matchType = matchNonFirst
	}
	matchType = strings.ToLower(matchType)
	matchFirstLine := false
	negateMatch := false
	switch matchType {
	case matchFirst:
		matchFirstLine = true
		negateMatch = false
	case matchLast:
		matchFirstLine = false
		negateMatch = false
	case matchNonFirst:
		matchFirstLine = true
		negateMatch = true
	case matchNonLast:
		matchFirstLine = false
		negateMatch = true
	default:
		return nil, errors.New("multiline: invalid value for MULTILINE_MATCH (must be one of first|last|nonfirst|nonlast): " + matchType)
	}

	flushAfter := 500 * time.Millisecond
	flushAfterStr := os.Getenv("MULTILINE_FLUSH_AFTER")
	if flushAfterStr != "" {
		timeoutMS, err := strconv.Atoi(flushAfterStr)
		if err != nil {
			return nil, errors.New("multiline: invalid value for multiline_timeout (must be number): " + flushAfterStr)
		}
		flushAfter = time.Duration(timeoutMS) * time.Millisecond
	}

	parts := strings.SplitN(route.Adapter, "+", 2)
	if len(parts) != 2 {
		return nil, errors.New("multiline: adapter must have a sub-adapter, eg: multiline+raw+tcp")
	}

	originalAdapter := route.Adapter
	route.Adapter = parts[1]
	factory, found := router.AdapterFactories.Lookup(route.AdapterType())
	if !found {
		return nil, errors.New("bad adapter: " + originalAdapter)
	}
	subAdapter, err := factory(route)
	if err != nil {
		return nil, err
	}
	route.Adapter = originalAdapter

	out := make(chan *router.Message)
	checkInterval := flushAfter / 2

	return &Adapter{
		out:             out,
		subAdapter:      subAdapter,
		enableByDefault: enableByDefault,
		pattern:         patternRegexp,
		separator:       separator,
		matchFirstLine:  matchFirstLine,
		negateMatch:     negateMatch,
		flushAfter:      flushAfter,
		checkInterval:   checkInterval,
		buffers:         make(map[string]*router.Message),
		nextCheck:       time.After(checkInterval),
	}, nil
}
1215783
func WeakMultiplex(output SimpleInChannel, inputs ...SimpleOutChannel) {
	if len(inputs) == 0 {
		panic("channels: WeakMultiplex requires at least one input")
	}
	go multiplex(output, inputs, false)
}
68996
func (sh *StreamHandler) Start() {
	sh.requests = make(map[*http.Request]chan []byte)
	sh.done = make(chan struct{})
	go sh.loop()
}
819350
func (s *NetlinkSocket) RequestMulti(req *NlMsgBuilder, consumer func(*NlMsgParser) error) error {
	seq, err := s.send(req)
	if err != nil {
		return err
	}

	return s.Receive(func(msg *NlMsgParser) (bool, error) {
		relevant, err := msg.checkResponseHeader(s.PortId(), seq)
		if !relevant || err != nil {
			return false, err
		}

		if msg.NlMsghdr().Type == syscall.NLMSG_DONE {
			return true, processNlMsgDone(msg)
		}

		err = consumer(msg)
		if err != nil {
			return true, err
		}

		return false, nil
	})
}
59667
func (p *BulkProcessor) Start(ctx context.Context) error {
	p.startedMu.Lock()
	defer p.startedMu.Unlock()

	if p.started {
		return nil
	}

	// We must have at least one worker.
	if p.numWorkers < 1 {
		p.numWorkers = 1
	}

	p.requestsC = make(chan BulkableRequest)
	p.executionId = 0
	p.stats = newBulkProcessorStats(p.numWorkers)
	p.stopReconnC = make(chan struct{})

	// Create and start up workers.
	p.workers = make([]*bulkWorker, p.numWorkers)
	for i := 0; i < p.numWorkers; i++ {
		p.workerWg.Add(1)
		p.workers[i] = newBulkWorker(p, i)
		go p.workers[i].work(ctx)
	}

	// Start the ticker for flush (if enabled)
	if int64(p.flushInterval) > 0 {
		p.flusherStopC = make(chan struct{})
		go p.flusher(p.flushInterval)
	}

	p.started = true

	return nil
}
58
func (co *Coordinator) StartPlugins() {
	// Launch routers
	for _, router := range co.routers {
		logrus.Debug("Starting ", reflect.TypeOf(router))
		if err := router.Start(); err != nil {
			logrus.WithError(err).Errorf("Failed to start router of type '%s'", reflect.TypeOf(router))
		}
	}

	// Launch producers
	co.state = coordinatorStateStartProducers
	for _, producer := range co.producers {
		producer := producer
		go tgo.WithRecoverShutdown(func() {
			logrus.Debug("Starting ", reflect.TypeOf(producer))
			producer.Produce(co.producerWorker)
		})
	}

	// Set final log target and purge the intermediate buffer
	if core.StreamRegistry.IsStreamRegistered(core.LogInternalStreamID) {
		// The _GOLLUM_ stream has listeners, so use LogConsumer to write to it
		if *flagLogColors == "always" {
			logrus.SetFormatter(logger.NewConsoleFormatter())
		}
		logrusHookBuffer.SetTargetHook(co.logConsumer)
		logrusHookBuffer.Purge()

	} else {
		logrusHookBuffer.SetTargetWriter(logger.FallbackLogDevice)
		logrusHookBuffer.Purge()
	}

	// Launch consumers
	co.state = coordinatorStateStartConsumers
	for _, consumer := range co.consumers {
		consumer := consumer
		go tgo.WithRecoverShutdown(func() {
			logrus.Debug("Starting ", reflect.TypeOf(consumer))
			consumer.Consume(co.consumerWorker)
		})
	}
}