// AddMultipleStreamMixService 往主流添加一路流
// {@link https://help.aliyun.com/document_detail/51315.html?spm=5176.doc51314.6.684.e6WbAe}
974081
func (l *Live) AddMultipleStreamMixService(mixStream MixStreamParam, resp interface{}) (err error) {
	req := l.cloneRequest(AddMultipleStreamMixServiceAction)
	if mixStream.Main.DomainName != "" {
		req.DomainName = mixStream.Main.DomainName
	}
	if mixStream.Main.AppName != "" {
		req.AppName = mixStream.Main.AppName
	}
	req.SetArgs("StreamName", mixStream.Main.StreamName)
	req.SetArgs("MixDomainName", mixStream.Mix.DomainName)
	req.SetArgs("MixAppName", mixStream.Mix.AppName)
	req.SetArgs("MixStreamName", mixStream.Mix.StreamName)
	err = l.rpc.Query(req, resp)
	return
}
881807
func (c *Client) SubscribeMultiple(subscriptions []packet.Subscription) (SubscribeFuture, error) {
	c.mutex.Lock()
	defer c.mutex.Unlock()

	// check if connected
	if atomic.LoadUint32(&c.state) != clientConnected {
		return nil, ErrClientNotConnected
	}

	// allocate subscribe packet
	subscribe := packet.NewSubscribe()
	subscribe.ID = c.Session.NextID()
	subscribe.Subscriptions = subscriptions

	// create future
	subFuture := future.New()

	// store future
	c.futureStore.Put(subscribe.ID, subFuture)

	// send packet
	err := c.send(subscribe, true)
	if err != nil {
		return nil, c.cleanup(err, false, false)
	}

	// wrap future
	wrappedFuture := &subscribeFuture{subFuture}

	return wrappedFuture, nil
}
767163
func (ens *NetpluginNameServer) AddLbService(tenant string, name string, v4name string) {
	if len(v4name) > 0 {
		dnsLog.Infof("add k8s service %s ip %s", name, v4name)
		if tenant == K8sDefaultTenant {
			ens.k8sService.Set(name, nameRecord{v4Record: net.ParseIP(v4name)})
		} else {
			mc := mastercfg.CfgServiceLBState{
				Tenant:      tenant,
				ServiceName: name,
				IPAddress:   v4name,
			}
			ens.addService(&mc)
		}
	}
}
310483
func addImageStreamLayersFromCache(isl *imageapi.ImageStreamLayers, is *imageapi.ImageStream, index ImageLayerIndex) []string {
	var missing []string
	for _, status := range is.Status.Tags {
		for _, item := range status.Items {
			if len(item.Image) == 0 {
				continue
			}

			obj, _, _ := index.GetByKey(item.Image)
			entry, ok := obj.(*ImageLayers)
			if !ok {
				if _, ok := isl.Images[item.Image]; !ok {
					isl.Images[item.Image] = imageapi.ImageBlobReferences{ImageMissing: true}
				}
				missing = append(missing, item.Image)
				continue
			}

			// we have already added this image once
			if _, ok := isl.Images[item.Image]; ok {
				continue
			}

			var reference imageapi.ImageBlobReferences
			for _, layer := range entry.Layers {
				reference.Layers = append(reference.Layers, layer.Name)
				if _, ok := isl.Blobs[layer.Name]; !ok {
					isl.Blobs[layer.Name] = imageapi.ImageLayerData{LayerSize: &layer.LayerSize, MediaType: layer.MediaType}
				}
			}

			if blob := entry.Config; blob != nil {
				reference.Config = &blob.Name
				if _, ok := isl.Blobs[blob.Name]; !ok {
					if blob.LayerSize == 0 {
						// only send media type since we don't the size of the manifest
						isl.Blobs[blob.Name] = imageapi.ImageLayerData{MediaType: blob.MediaType}
					} else {
						isl.Blobs[blob.Name] = imageapi.ImageLayerData{LayerSize: &blob.LayerSize, MediaType: blob.MediaType}
					}
				}
			}

			// the image manifest is always a blob - schema2 images also have a config blob referenced from the manifest
			if _, ok := isl.Blobs[item.Image]; !ok {
				isl.Blobs[item.Image] = imageapi.ImageLayerData{MediaType: entry.MediaType}
			}
			isl.Images[item.Image] = reference
		}
	}
	return missing
}
248248
func (f *File) addMedia(file []byte, ext string) string {
	count := f.countMedia()
	for name, existing := range f.XLSX {
		if !strings.HasPrefix(name, "xl/media/image") {
			continue
		}
		if bytes.Equal(file, existing) {
			return name
		}
	}
	media := "xl/media/image" + strconv.Itoa(count+1) + ext
	f.XLSX[media] = file
	return media
}
767904
func (jf openTracingService) AddGrpcServerOptions(addInterceptors func(s grpc.StreamServerInterceptor, u grpc.UnaryServerInterceptor)) {
	addInterceptors(otgrpc.OpenTracingStreamServerInterceptor(jf.Tracer), otgrpc.OpenTracingServerInterceptor(jf.Tracer))
}
1103317
func NewMixProvider(first logger.Provider, others ...logger.Provider) logger.Provider {
	p := new(mixProvider)
	p.providers = make([]logger.Provider, 0, len(others)+1)
	p.providers = append(p.providers, first)
	for _, other := range others {
		p.providers = append(p.providers, other)
	}
	return p
}
1232742
func (l *Live) AddLivePullStreamInfoConfig(streamName, sourceUrl string, startTime, endTime time.Time, resp interface{}) (err error) {
	req := l.cloneRequest(AddLivePullStreamInfoConfigAction)
	req.SetArgs("StreamName", streamName)
	req.SetArgs("SourceUrl", sourceUrl)
	req.SetArgs("StartTime", util.GetISO8601TimeStamp(startTime))
	req.SetArgs("EndTime", util.GetISO8601TimeStamp(endTime))
	err = l.rpc.Query(req, resp)
	return
}
119348
func NewPlusDmForStreamWithSrcLen(sourceLength uint, priceStream gotrade.DOHLCVStreamSubscriber, timePeriod int) (indicator *PlusDm, err error) {
	ind, err := NewPlusDmWithSrcLen(sourceLength, timePeriod)
	priceStream.AddTickSubscription(ind)
	return ind, err
}
884660
func (cfs *cacheFSObjects) NewMultipartUpload(ctx context.Context, bucket, object string, uploadID string, opts ObjectOptions) (string, error) {
	if cfs.diskUsageHigh() {
		select {
		case cfs.purgeChan <- struct{}{}:
		default:
		}
		return "", errDiskFull
	}

	if _, err := cfs.GetBucketInfo(ctx, bucket); err != nil {
		pErr := cfs.MakeBucketWithLocation(ctx, bucket, "")
		if pErr != nil {
			return "", pErr
		}
	}
	fs := cfs.FSObjects
	if err := checkNewMultipartArgs(ctx, bucket, object, fs); err != nil {
		return "", toObjectErr(err, bucket)
	}

	if _, err := fs.statBucketDir(ctx, bucket); err != nil {
		return "", toObjectErr(err, bucket)
	}

	uploadIDDir := fs.getUploadIDDir(bucket, object, uploadID)

	err := mkdirAll(uploadIDDir, 0755)
	if err != nil {
		logger.LogIf(ctx, err)
		return "", err
	}

	// Initialize fs.json values.
	fsMeta := newFSMetaV1()
	fsMeta.Meta = opts.UserDefined

	fsMetaBytes, err := json.Marshal(fsMeta)
	if err != nil {
		logger.LogIf(ctx, err)
		return "", err
	}

	if err = ioutil.WriteFile(pathJoin(uploadIDDir, fs.metaJSONFile), fsMetaBytes, 0644); err != nil {
		logger.LogIf(ctx, err)
		return "", err
	}
	return uploadID, nil
}
855869
func (a *azureObjects) NewMultipartUpload(ctx context.Context, bucket, object string, opts minio.ObjectOptions) (uploadID string, err error) {
	uploadID, err = getAzureUploadID()
	if err != nil {
		logger.LogIf(ctx, err)
		return "", err
	}
	metadataObject := getAzureMetadataObjectName(object, uploadID)

	var jsonData []byte
	if jsonData, err = json.Marshal(azureMultipartMetadata{Name: object, Metadata: opts.UserDefined}); err != nil {
		logger.LogIf(ctx, err)
		return "", err
	}

	blob := a.client.GetContainerReference(bucket).GetBlobReference(metadataObject)
	err = blob.CreateBlockBlobFromReader(bytes.NewBuffer(jsonData), nil)
	if err != nil {
		return "", azureToObjectError(err, bucket, metadataObject)
	}

	return uploadID, nil
}
185504
func (c *HyperClient) AddService(podID string, services []*types.UserService) error {
	_, err := c.client.ServiceAdd(
		c.ctx,
		&types.ServiceAddRequest{PodID: podID, Services: services},
	)

	if err != nil {
		return err
	}

	return nil
}
188531
func callPostMixFunction(udata unsafe.Pointer, stream *C.Uint8, length C.int) {
	var slice []uint8
	header := (*reflect.SliceHeader)(unsafe.Pointer(&slice))
	header.Data = uintptr(unsafe.Pointer(stream))
	header.Len = int(length)
	header.Cap = int(length)
	postMixFunc(slice)
}
1098630
func (api *ProductDiskAPI) SetFilterMultiBy(key string, value interface{}) {
	api.filterBy(key, value, true)
}
1195782
func AddStreamInterceptor(sint grpc.StreamServerInterceptor) {
	service.StreamInts = append(service.StreamInts, sint)
}
205058
func (l *s3EncObjects) NewMultipartUpload(ctx context.Context, bucket string, object string, o minio.ObjectOptions) (uploadID string, err error) {
	var sseOpts encrypt.ServerSide
	if o.ServerSideEncryption == nil {
		return l.s3Objects.NewMultipartUpload(ctx, bucket, object, minio.ObjectOptions{UserDefined: o.UserDefined})
	}
	// Decide if sse options needed to be passed to backend
	if (minio.GlobalGatewaySSE.SSEC() && o.ServerSideEncryption.Type() == encrypt.SSEC) ||
		(minio.GlobalGatewaySSE.SSES3() && o.ServerSideEncryption.Type() == encrypt.S3) {
		sseOpts = o.ServerSideEncryption
	}

	uploadID, err = l.s3Objects.NewMultipartUpload(ctx, bucket, getGWContentPath(object), minio.ObjectOptions{ServerSideEncryption: sseOpts})
	if err != nil {
		return
	}
	// Create uploadID and write a temporary dare.meta object under object/uploadID prefix
	gwmeta := newGWMetaV1()
	gwmeta.Meta = o.UserDefined
	gwmeta.Stat.ModTime = time.Now().UTC()
	err = l.writeGWMetadata(ctx, bucket, getTmpDareMetaPath(object, uploadID), gwmeta, minio.ObjectOptions{})
	if err != nil {
		return uploadID, minio.ErrorRespToObjectError(err)
	}
	return uploadID, nil
}
1031218
func NewMultistream(c io.ReadWriteCloser, proto string) Multistream {
	return &lazyClientConn{
		protos: []string{proto},
		con:    c,
	}
}