mongodbatlas.getStreamProcessors

import * as pulumi from "@pulumi/pulumi";
import * as mongodbatlas from "@pulumi/mongodbatlas";

const example = new mongodbatlas.StreamInstance("example", {
    projectId: projectId,
    instanceName: "InstanceName",
    dataProcessRegion: {
        region: "VIRGINIA_USA",
        cloudProvider: "AWS",
    },
});
const example_sample = new mongodbatlas.StreamConnection("example-sample", {
    projectId: projectId,
    workspaceName: example.instanceName,
    connectionName: "sample_stream_solar",
    type: "Sample",
});
const example_cluster = new mongodbatlas.StreamConnection("example-cluster", {
    projectId: projectId,
    workspaceName: example.instanceName,
    connectionName: "ClusterConnection",
    type: "Cluster",
    clusterName: clusterName,
    dbRoleToExecute: {
        role: "atlasAdmin",
        type: "BUILT_IN",
    },
});
const example_kafka = new mongodbatlas.StreamConnection("example-kafka", {
    projectId: projectId,
    workspaceName: example.instanceName,
    connectionName: "KafkaPlaintextConnection",
    type: "Kafka",
    authentication: {
        mechanism: "PLAIN",
        username: kafkaUsername,
        password: kafkaPassword,
    },
    bootstrapServers: "example.com:9092",
    config: {
        "auto.offset.reset": "earliest",
    },
    security: {
        protocol: "SASL_PLAINTEXT",
    },
});
const stream_processor_sample_example = new mongodbatlas.StreamProcessor("stream-processor-sample-example", {
    projectId: projectId,
    workspaceName: example.instanceName,
    processorName: "sampleProcessorName",
    pipeline: pulumi.interpolate`[
  {\"$source\": {\"connectionName\": \"${example_sample.connectionName}\"}},
  {\"$emit\": {\"connectionName\": \"${example_cluster.connectionName}\", \"db\": \"sample\", \"coll\": \"solar\", \"timeseries\": {\"timeField\": \"_ts\"}}}
]
`,
    state: "STARTED",
    tier: "SP30",
});
const stream_processor_cluster_to_kafka_example = new mongodbatlas.StreamProcessor("stream-processor-cluster-to-kafka-example", {
    projectId: projectId,
    workspaceName: example.instanceName,
    processorName: "clusterProcessorName",
    pipeline: pulumi.interpolate`[
  {\"$source\": {\"connectionName\": \"${example_cluster.connectionName}\"}},
  {\"$emit\": {\"connectionName\": \"${example_kafka.connectionName}\", \"topic\": \"topic_from_cluster\"}}
]
`,
    state: "CREATED",
});
const stream_processor_kafka_to_cluster_example = new mongodbatlas.StreamProcessor("stream-processor-kafka-to-cluster-example", {
    projectId: projectId,
    workspaceName: example.instanceName,
    processorName: "kafkaProcessorName",
    pipeline: pulumi.interpolate`[
  {\"$source\": {\"connectionName\": \"${example_kafka.connectionName}\", \"topic\": \"topic_source\"}},
  {\"$emit\": {\"connectionName\": \"${example_cluster.connectionName}\", \"db\": \"kafka\", \"coll\": \"topic_source\", \"timeseries\": {\"timeField\": \"ts\"}}}
]
`,
    state: "CREATED",
    tier: "SP10",
    options: {
        dlq: {
            coll: "exampleColumn",
            connectionName: example_cluster.connectionName,
            db: "exampleDb",
        },
        autoscaling: {
            minTier: "SP10",
            maxTier: "SP50",
        },
    },
});
const example_stream_processors = mongodbatlas.getStreamProcessorsOutput({
    projectId: projectId,
    workspaceName: example.instanceName,
});
const example_stream_processor = mongodbatlas.getStreamProcessorOutput({
    projectId: projectId,
    workspaceName: example.instanceName,
    processorName: stream_processor_sample_example.processorName,
});
export const streamProcessorsState = example_stream_processor.state;
export const streamProcessorsResults = example_stream_processors.apply(example_stream_processors => example_stream_processors.results);

import pulumi
import pulumi_mongodbatlas as mongodbatlas

example = mongodbatlas.StreamInstance("example",
    project_id=project_id,
    instance_name="InstanceName",
    data_process_region={
        "region": "VIRGINIA_USA",
        "cloud_provider": "AWS",
    })
example_sample = mongodbatlas.StreamConnection("example-sample",
    project_id=project_id,
    workspace_name=example.instance_name,
    connection_name="sample_stream_solar",
    type="Sample")
example_cluster = mongodbatlas.StreamConnection("example-cluster",
    project_id=project_id,
    workspace_name=example.instance_name,
    connection_name="ClusterConnection",
    type="Cluster",
    cluster_name=cluster_name,
    db_role_to_execute={
        "role": "atlasAdmin",
        "type": "BUILT_IN",
    })
example_kafka = mongodbatlas.StreamConnection("example-kafka",
    project_id=project_id,
    workspace_name=example.instance_name,
    connection_name="KafkaPlaintextConnection",
    type="Kafka",
    authentication={
        "mechanism": "PLAIN",
        "username": kafka_username,
        "password": kafka_password,
    },
    bootstrap_servers="example.com:9092",
    config={
        "auto.offset.reset": "earliest",
    },
    security={
        "protocol": "SASL_PLAINTEXT",
    })
stream_processor_sample_example = mongodbatlas.StreamProcessor("stream-processor-sample-example",
    project_id=project_id,
    workspace_name=example.instance_name,
    processor_name="sampleProcessorName",
    pipeline=pulumi.Output.all(
        example-sampleConnection_name=example_sample.connection_name,
        example-clusterConnection_name=example_cluster.connection_name
).apply(lambda resolved_outputs: f"""[
  {{\"$source\": {{\"connectionName\": \"{resolved_outputs['example-sampleConnection_name']}\"}}}},
  {{\"$emit\": {{\"connectionName\": \"{resolved_outputs['example-clusterConnection_name']}\", \"db\": \"sample\", \"coll\": \"solar\", \"timeseries\": {{\"timeField\": \"_ts\"}}}}}}
]
""")
,
    state="STARTED",
    tier="SP30")
stream_processor_cluster_to_kafka_example = mongodbatlas.StreamProcessor("stream-processor-cluster-to-kafka-example",
    project_id=project_id,
    workspace_name=example.instance_name,
    processor_name="clusterProcessorName",
    pipeline=pulumi.Output.all(
        example-clusterConnection_name=example_cluster.connection_name,
        example-kafkaConnection_name=example_kafka.connection_name
).apply(lambda resolved_outputs: f"""[
  {{\"$source\": {{\"connectionName\": \"{resolved_outputs['example-clusterConnection_name']}\"}}}},
  {{\"$emit\": {{\"connectionName\": \"{resolved_outputs['example-kafkaConnection_name']}\", \"topic\": \"topic_from_cluster\"}}}}
]
""")
,
    state="CREATED")
stream_processor_kafka_to_cluster_example = mongodbatlas.StreamProcessor("stream-processor-kafka-to-cluster-example",
    project_id=project_id,
    workspace_name=example.instance_name,
    processor_name="kafkaProcessorName",
    pipeline=pulumi.Output.all(
        example-kafkaConnection_name=example_kafka.connection_name,
        example-clusterConnection_name=example_cluster.connection_name
).apply(lambda resolved_outputs: f"""[
  {{\"$source\": {{\"connectionName\": \"{resolved_outputs['example-kafkaConnection_name']}\", \"topic\": \"topic_source\"}}}},
  {{\"$emit\": {{\"connectionName\": \"{resolved_outputs['example-clusterConnection_name']}\", \"db\": \"kafka\", \"coll\": \"topic_source\", \"timeseries\": {{\"timeField\": \"ts\"}}}}}}
]
""")
,
    state="CREATED",
    tier="SP10",
    options={
        "dlq": {
            "coll": "exampleColumn",
            "connection_name": example_cluster.connection_name,
            "db": "exampleDb",
        },
        "autoscaling": {
            "min_tier": "SP10",
            "max_tier": "SP50",
        },
    })
example_stream_processors = mongodbatlas.get_stream_processors_output(project_id=project_id,
    workspace_name=example.instance_name)
example_stream_processor = mongodbatlas.get_stream_processor_output(project_id=project_id,
    workspace_name=example.instance_name,
    processor_name=stream_processor_sample_example.processor_name)
pulumi.export("streamProcessorsState", example_stream_processor.state)
pulumi.export("streamProcessorsResults", example_stream_processors.results)

package main

import (
	"fmt"

	"github.com/pulumi/pulumi-mongodbatlas/sdk/v4/go/mongodbatlas"
	"github.com/pulumi/pulumi/sdk/v3/go/pulumi"
)

func main() {
	pulumi.Run(func(ctx *pulumi.Context) error {
		example, err := mongodbatlas.NewStreamInstance(ctx, "example", &mongodbatlas.StreamInstanceArgs{
			ProjectId:    pulumi.Any(projectId),
			InstanceName: pulumi.String("InstanceName"),
			DataProcessRegion: &mongodbatlas.StreamInstanceDataProcessRegionArgs{
				Region:        pulumi.String("VIRGINIA_USA"),
				CloudProvider: pulumi.String("AWS"),
			},
		})
		if err != nil {
			return err
		}
		example_sample, err := mongodbatlas.NewStreamConnection(ctx, "example-sample", &mongodbatlas.StreamConnectionArgs{
			ProjectId:      pulumi.Any(projectId),
			WorkspaceName:  example.InstanceName,
			ConnectionName: pulumi.String("sample_stream_solar"),
			Type:           pulumi.String("Sample"),
		})
		if err != nil {
			return err
		}
		example_cluster, err := mongodbatlas.NewStreamConnection(ctx, "example-cluster", &mongodbatlas.StreamConnectionArgs{
			ProjectId:      pulumi.Any(projectId),
			WorkspaceName:  example.InstanceName,
			ConnectionName: pulumi.String("ClusterConnection"),
			Type:           pulumi.String("Cluster"),
			ClusterName:    pulumi.Any(clusterName),
			DbRoleToExecute: &mongodbatlas.StreamConnectionDbRoleToExecuteArgs{
				Role: pulumi.String("atlasAdmin"),
				Type: pulumi.String("BUILT_IN"),
			},
		})
		if err != nil {
			return err
		}
		example_kafka, err := mongodbatlas.NewStreamConnection(ctx, "example-kafka", &mongodbatlas.StreamConnectionArgs{
			ProjectId:      pulumi.Any(projectId),
			WorkspaceName:  example.InstanceName,
			ConnectionName: pulumi.String("KafkaPlaintextConnection"),
			Type:           pulumi.String("Kafka"),
			Authentication: &mongodbatlas.StreamConnectionAuthenticationArgs{
				Mechanism: pulumi.String("PLAIN"),
				Username:  pulumi.Any(kafkaUsername),
				Password:  pulumi.Any(kafkaPassword),
			},
			BootstrapServers: pulumi.String("example.com:9092"),
			Config: pulumi.StringMap{
				"auto.offset.reset": pulumi.String("earliest"),
			},
			Security: &mongodbatlas.StreamConnectionSecurityArgs{
				Protocol: pulumi.String("SASL_PLAINTEXT"),
			},
		})
		if err != nil {
			return err
		}
		stream_processor_sample_example, err := mongodbatlas.NewStreamProcessor(ctx, "stream-processor-sample-example", &mongodbatlas.StreamProcessorArgs{
			ProjectId:     pulumi.Any(projectId),
			WorkspaceName: example.InstanceName,
			ProcessorName: pulumi.String("sampleProcessorName"),
			Pipeline: pulumi.All(example_sample.ConnectionName, example_cluster.ConnectionName).ApplyT(func(_args []interface{}) (string, error) {
				example - sampleConnectionName := _args[0].(string)
				example - clusterConnectionName := _args[1].(string)
				return fmt.Sprintf("[\n  {\\\"$source\\\": {\\\"connectionName\\\": \\\"%v\\\"}},\n  {\\\"$emit\\\": {\\\"connectionName\\\": \\\"%v\\\", \\\"db\\\": \\\"sample\\\", \\\"coll\\\": \\\"solar\\\", \\\"timeseries\\\": {\\\"timeField\\\": \\\"_ts\\\"}}}\n]\n", example_sampleConnectionName, example_clusterConnectionName), nil
			}).(pulumi.StringOutput),
			State: pulumi.String("STARTED"),
			Tier:  pulumi.String("SP30"),
		})
		if err != nil {
			return err
		}
		_, err = mongodbatlas.NewStreamProcessor(ctx, "stream-processor-cluster-to-kafka-example", &mongodbatlas.StreamProcessorArgs{
			ProjectId:     pulumi.Any(projectId),
			WorkspaceName: example.InstanceName,
			ProcessorName: pulumi.String("clusterProcessorName"),
			Pipeline: pulumi.All(example_cluster.ConnectionName, example_kafka.ConnectionName).ApplyT(func(_args []interface{}) (string, error) {
				example - clusterConnectionName := _args[0].(string)
				example - kafkaConnectionName := _args[1].(string)
				return fmt.Sprintf("[\n  {\\\"$source\\\": {\\\"connectionName\\\": \\\"%v\\\"}},\n  {\\\"$emit\\\": {\\\"connectionName\\\": \\\"%v\\\", \\\"topic\\\": \\\"topic_from_cluster\\\"}}\n]\n", example_clusterConnectionName, example_kafkaConnectionName), nil
			}).(pulumi.StringOutput),
			State: pulumi.String("CREATED"),
		})
		if err != nil {
			return err
		}
		_, err = mongodbatlas.NewStreamProcessor(ctx, "stream-processor-kafka-to-cluster-example", &mongodbatlas.StreamProcessorArgs{
			ProjectId:     pulumi.Any(projectId),
			WorkspaceName: example.InstanceName,
			ProcessorName: pulumi.String("kafkaProcessorName"),
			Pipeline: pulumi.All(example_kafka.ConnectionName, example_cluster.ConnectionName).ApplyT(func(_args []interface{}) (string, error) {
				example - kafkaConnectionName := _args[0].(string)
				example - clusterConnectionName := _args[1].(string)
				return fmt.Sprintf("[\n  {\\\"$source\\\": {\\\"connectionName\\\": \\\"%v\\\", \\\"topic\\\": \\\"topic_source\\\"}},\n  {\\\"$emit\\\": {\\\"connectionName\\\": \\\"%v\\\", \\\"db\\\": \\\"kafka\\\", \\\"coll\\\": \\\"topic_source\\\", \\\"timeseries\\\": {\\\"timeField\\\": \\\"ts\\\"}}}\n]\n", example_kafkaConnectionName, example_clusterConnectionName), nil
			}).(pulumi.StringOutput),
			State: pulumi.String("CREATED"),
			Tier:  pulumi.String("SP10"),
			Options: &mongodbatlas.StreamProcessorOptionsArgs{
				Dlq: &mongodbatlas.StreamProcessorOptionsDlqArgs{
					Coll:           pulumi.String("exampleColumn"),
					ConnectionName: example_cluster.ConnectionName,
					Db:             pulumi.String("exampleDb"),
				},
				Autoscaling: &mongodbatlas.StreamProcessorOptionsAutoscalingArgs{
					MinTier: pulumi.String("SP10"),
					MaxTier: pulumi.String("SP50"),
				},
			},
		})
		if err != nil {
			return err
		}
		example_stream_processors := mongodbatlas.GetStreamProcessorsOutput(ctx, mongodbatlas.GetStreamProcessorsOutputArgs{
			ProjectId:     pulumi.Any(projectId),
			WorkspaceName: example.InstanceName,
		}, nil)
		example_stream_processor := mongodbatlas.GetStreamProcessorOutput(ctx, mongodbatlas.GetStreamProcessorOutputArgs{
			ProjectId:     pulumi.Any(projectId),
			WorkspaceName: example.InstanceName,
			ProcessorName: stream_processor_sample_example.ProcessorName,
		}, nil)
		ctx.Export("streamProcessorsState", example_stream_processor.State())
		ctx.Export("streamProcessorsResults", example_stream_processors.ApplyT(func(example_stream_processors mongodbatlas.GetStreamProcessorsResult) ([]mongodbatlas.GetStreamProcessorsResult, error) {
			return example_stream_processors.Results.([]mongodbatlas.GetStreamProcessorsResult), nil
		}).(pulumi.ArrayOutput))
		return nil
	})
}

using System.Collections.Generic;
using System.Linq;
using Pulumi;
using Mongodbatlas = Pulumi.Mongodbatlas;

return await Deployment.RunAsync(() => 
{
    var example = new Mongodbatlas.StreamInstance("example", new()
    {
        ProjectId = projectId,
        InstanceName = "InstanceName",
        DataProcessRegion = new Mongodbatlas.Inputs.StreamInstanceDataProcessRegionArgs
        {
            Region = "VIRGINIA_USA",
            CloudProvider = "AWS",
        },
    });

    var example_sample = new Mongodbatlas.StreamConnection("example-sample", new()
    {
        ProjectId = projectId,
        WorkspaceName = example.InstanceName,
        ConnectionName = "sample_stream_solar",
        Type = "Sample",
    });

    var example_cluster = new Mongodbatlas.StreamConnection("example-cluster", new()
    {
        ProjectId = projectId,
        WorkspaceName = example.InstanceName,
        ConnectionName = "ClusterConnection",
        Type = "Cluster",
        ClusterName = clusterName,
        DbRoleToExecute = new Mongodbatlas.Inputs.StreamConnectionDbRoleToExecuteArgs
        {
            Role = "atlasAdmin",
            Type = "BUILT_IN",
        },
    });

    var example_kafka = new Mongodbatlas.StreamConnection("example-kafka", new()
    {
        ProjectId = projectId,
        WorkspaceName = example.InstanceName,
        ConnectionName = "KafkaPlaintextConnection",
        Type = "Kafka",
        Authentication = new Mongodbatlas.Inputs.StreamConnectionAuthenticationArgs
        {
            Mechanism = "PLAIN",
            Username = kafkaUsername,
            Password = kafkaPassword,
        },
        BootstrapServers = "example.com:9092",
        Config = 
        {
            { "auto.offset.reset", "earliest" },
        },
        Security = new Mongodbatlas.Inputs.StreamConnectionSecurityArgs
        {
            Protocol = "SASL_PLAINTEXT",
        },
    });

    var stream_processor_sample_example = new Mongodbatlas.StreamProcessor("stream-processor-sample-example", new()
    {
        ProjectId = projectId,
        WorkspaceName = example.InstanceName,
        ProcessorName = "sampleProcessorName",
        Pipeline = Output.Tuple(example_sample.ConnectionName, example_cluster.ConnectionName).Apply(values =>
        {
            var example-sampleConnectionName = values.Item1;
            var example-clusterConnectionName = values.Item2;
            return @$"[
  {{\""$source\"": {{\""connectionName\"": \""{example_sampleConnectionName}\""}}}},
  {{\""$emit\"": {{\""connectionName\"": \""{example_clusterConnectionName}\"", \""db\"": \""sample\"", \""coll\"": \""solar\"", \""timeseries\"": {{\""timeField\"": \""_ts\""}}}}}}
]
";
        }),
        State = "STARTED",
        Tier = "SP30",
    });

    var stream_processor_cluster_to_kafka_example = new Mongodbatlas.StreamProcessor("stream-processor-cluster-to-kafka-example", new()
    {
        ProjectId = projectId,
        WorkspaceName = example.InstanceName,
        ProcessorName = "clusterProcessorName",
        Pipeline = Output.Tuple(example_cluster.ConnectionName, example_kafka.ConnectionName).Apply(values =>
        {
            var example-clusterConnectionName = values.Item1;
            var example-kafkaConnectionName = values.Item2;
            return @$"[
  {{\""$source\"": {{\""connectionName\"": \""{example_clusterConnectionName}\""}}}},
  {{\""$emit\"": {{\""connectionName\"": \""{example_kafkaConnectionName}\"", \""topic\"": \""topic_from_cluster\""}}}}
]
";
        }),
        State = "CREATED",
    });

    var stream_processor_kafka_to_cluster_example = new Mongodbatlas.StreamProcessor("stream-processor-kafka-to-cluster-example", new()
    {
        ProjectId = projectId,
        WorkspaceName = example.InstanceName,
        ProcessorName = "kafkaProcessorName",
        Pipeline = Output.Tuple(example_kafka.ConnectionName, example_cluster.ConnectionName).Apply(values =>
        {
            var example-kafkaConnectionName = values.Item1;
            var example-clusterConnectionName = values.Item2;
            return @$"[
  {{\""$source\"": {{\""connectionName\"": \""{example_kafkaConnectionName}\"", \""topic\"": \""topic_source\""}}}},
  {{\""$emit\"": {{\""connectionName\"": \""{example_clusterConnectionName}\"", \""db\"": \""kafka\"", \""coll\"": \""topic_source\"", \""timeseries\"": {{\""timeField\"": \""ts\""}}}}}}
]
";
        }),
        State = "CREATED",
        Tier = "SP10",
        Options = new Mongodbatlas.Inputs.StreamProcessorOptionsArgs
        {
            Dlq = new Mongodbatlas.Inputs.StreamProcessorOptionsDlqArgs
            {
                Coll = "exampleColumn",
                ConnectionName = example_cluster.ConnectionName,
                Db = "exampleDb",
            },
            Autoscaling = new Mongodbatlas.Inputs.StreamProcessorOptionsAutoscalingArgs
            {
                MinTier = "SP10",
                MaxTier = "SP50",
            },
        },
    });

    var example_stream_processors = Mongodbatlas.GetStreamProcessors.Invoke(new()
    {
        ProjectId = projectId,
        WorkspaceName = example.InstanceName,
    });

    var example_stream_processor = Mongodbatlas.GetStreamProcessor.Invoke(new()
    {
        ProjectId = projectId,
        WorkspaceName = example.InstanceName,
        ProcessorName = stream_processor_sample_example.ProcessorName,
    });

    return new Dictionary<string, object?>
    {
        ["streamProcessorsState"] = example_stream_processor.Apply(example_stream_processor => example_stream_processor.Apply(getStreamProcessorResult => getStreamProcessorResult.State)),
        ["streamProcessorsResults"] = example_stream_processors.Apply(example_stream_processors => example_stream_processors.Apply(getStreamProcessorsResult => getStreamProcessorsResult.Results)),
    };
});

package generated_program;

import com.pulumi.Context;
import com.pulumi.Pulumi;
import com.pulumi.core.Output;
import com.pulumi.mongodbatlas.StreamInstance;
import com.pulumi.mongodbatlas.StreamInstanceArgs;
import com.pulumi.mongodbatlas.inputs.StreamInstanceDataProcessRegionArgs;
import com.pulumi.mongodbatlas.StreamConnection;
import com.pulumi.mongodbatlas.StreamConnectionArgs;
import com.pulumi.mongodbatlas.inputs.StreamConnectionDbRoleToExecuteArgs;
import com.pulumi.mongodbatlas.inputs.StreamConnectionAuthenticationArgs;
import com.pulumi.mongodbatlas.inputs.StreamConnectionSecurityArgs;
import com.pulumi.mongodbatlas.StreamProcessor;
import com.pulumi.mongodbatlas.StreamProcessorArgs;
import com.pulumi.mongodbatlas.inputs.StreamProcessorOptionsArgs;
import com.pulumi.mongodbatlas.inputs.StreamProcessorOptionsDlqArgs;
import com.pulumi.mongodbatlas.inputs.StreamProcessorOptionsAutoscalingArgs;
import com.pulumi.mongodbatlas.MongodbatlasFunctions;
import com.pulumi.mongodbatlas.inputs.GetStreamProcessorsArgs;
import com.pulumi.mongodbatlas.inputs.GetStreamProcessorArgs;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Map;
import java.io.File;
import java.nio.file.Files;
import java.nio.file.Paths;

public class App {
    public static void main(String[] args) {
        Pulumi.run(App::stack);
    }

    public static void stack(Context ctx) {
        var example = new StreamInstance("example", StreamInstanceArgs.builder()
            .projectId(projectId)
            .instanceName("InstanceName")
            .dataProcessRegion(StreamInstanceDataProcessRegionArgs.builder()
                .region("VIRGINIA_USA")
                .cloudProvider("AWS")
                .build())
            .build());

        var example_sample = new StreamConnection("example-sample", StreamConnectionArgs.builder()
            .projectId(projectId)
            .workspaceName(example.instanceName())
            .connectionName("sample_stream_solar")
            .type("Sample")
            .build());

        var example_cluster = new StreamConnection("example-cluster", StreamConnectionArgs.builder()
            .projectId(projectId)
            .workspaceName(example.instanceName())
            .connectionName("ClusterConnection")
            .type("Cluster")
            .clusterName(clusterName)
            .dbRoleToExecute(StreamConnectionDbRoleToExecuteArgs.builder()
                .role("atlasAdmin")
                .type("BUILT_IN")
                .build())
            .build());

        var example_kafka = new StreamConnection("example-kafka", StreamConnectionArgs.builder()
            .projectId(projectId)
            .workspaceName(example.instanceName())
            .connectionName("KafkaPlaintextConnection")
            .type("Kafka")
            .authentication(StreamConnectionAuthenticationArgs.builder()
                .mechanism("PLAIN")
                .username(kafkaUsername)
                .password(kafkaPassword)
                .build())
            .bootstrapServers("example.com:9092")
            .config(Map.of("auto.offset.reset", "earliest"))
            .security(StreamConnectionSecurityArgs.builder()
                .protocol("SASL_PLAINTEXT")
                .build())
            .build());

        var stream_processor_sample_example = new StreamProcessor("stream-processor-sample-example", StreamProcessorArgs.builder()
            .projectId(projectId)
            .workspaceName(example.instanceName())
            .processorName("sampleProcessorName")
            .pipeline(Output.tuple(example_sample.connectionName(), example_cluster.connectionName()).applyValue(values -> {
                var example-sampleConnectionName = values.t1;
                var example-clusterConnectionName = values.t2;
                return """
[
  {\"$source\": {\"connectionName\": \"%s\"}},
  {\"$emit\": {\"connectionName\": \"%s\", \"db\": \"sample\", \"coll\": \"solar\", \"timeseries\": {\"timeField\": \"_ts\"}}}
]
", example_sampleConnectionName,example_clusterConnectionName);
            }))
            .state("STARTED")
            .tier("SP30")
            .build());

        var stream_processor_cluster_to_kafka_example = new StreamProcessor("stream-processor-cluster-to-kafka-example", StreamProcessorArgs.builder()
            .projectId(projectId)
            .workspaceName(example.instanceName())
            .processorName("clusterProcessorName")
            .pipeline(Output.tuple(example_cluster.connectionName(), example_kafka.connectionName()).applyValue(values -> {
                var example-clusterConnectionName = values.t1;
                var example-kafkaConnectionName = values.t2;
                return """
[
  {\"$source\": {\"connectionName\": \"%s\"}},
  {\"$emit\": {\"connectionName\": \"%s\", \"topic\": \"topic_from_cluster\"}}
]
", example_clusterConnectionName,example_kafkaConnectionName);
            }))
            .state("CREATED")
            .build());

        var stream_processor_kafka_to_cluster_example = new StreamProcessor("stream-processor-kafka-to-cluster-example", StreamProcessorArgs.builder()
            .projectId(projectId)
            .workspaceName(example.instanceName())
            .processorName("kafkaProcessorName")
            .pipeline(Output.tuple(example_kafka.connectionName(), example_cluster.connectionName()).applyValue(values -> {
                var example-kafkaConnectionName = values.t1;
                var example-clusterConnectionName = values.t2;
                return """
[
  {\"$source\": {\"connectionName\": \"%s\", \"topic\": \"topic_source\"}},
  {\"$emit\": {\"connectionName\": \"%s\", \"db\": \"kafka\", \"coll\": \"topic_source\", \"timeseries\": {\"timeField\": \"ts\"}}}
]
", example_kafkaConnectionName,example_clusterConnectionName);
            }))
            .state("CREATED")
            .tier("SP10")
            .options(StreamProcessorOptionsArgs.builder()
                .dlq(StreamProcessorOptionsDlqArgs.builder()
                    .coll("exampleColumn")
                    .connectionName(example_cluster.connectionName())
                    .db("exampleDb")
                    .build())
                .autoscaling(StreamProcessorOptionsAutoscalingArgs.builder()
                    .minTier("SP10")
                    .maxTier("SP50")
                    .build())
                .build())
            .build());

        final var example-stream-processors = MongodbatlasFunctions.getStreamProcessors(GetStreamProcessorsArgs.builder()
            .projectId(projectId)
            .workspaceName(example.instanceName())
            .build());

        final var example-stream-processor = MongodbatlasFunctions.getStreamProcessor(GetStreamProcessorArgs.builder()
            .projectId(projectId)
            .workspaceName(example.instanceName())
            .processorName(stream_processor_sample_example.processorName())
            .build());

        ctx.export("streamProcessorsState", example_stream_processor.applyValue(_example_stream_processor -> _example_stream_processor.state()));
        ctx.export("streamProcessorsResults", example_stream_processors.applyValue(_example_stream_processors -> _example_stream_processors.results()));
    }
}

resources:
  example:
    type: mongodbatlas:StreamInstance
    properties:
      projectId: ${projectId}
      instanceName: InstanceName
      dataProcessRegion:
        region: VIRGINIA_USA
        cloudProvider: AWS
  example-sample:
    type: mongodbatlas:StreamConnection
    properties:
      projectId: ${projectId}
      workspaceName: ${example.instanceName}
      connectionName: sample_stream_solar
      type: Sample
  example-cluster:
    type: mongodbatlas:StreamConnection
    properties:
      projectId: ${projectId}
      workspaceName: ${example.instanceName}
      connectionName: ClusterConnection
      type: Cluster
      clusterName: ${clusterName}
      dbRoleToExecute:
        role: atlasAdmin
        type: BUILT_IN
  example-kafka:
    type: mongodbatlas:StreamConnection
    properties:
      projectId: ${projectId}
      workspaceName: ${example.instanceName}
      connectionName: KafkaPlaintextConnection
      type: Kafka
      authentication:
        mechanism: PLAIN
        username: ${kafkaUsername}
        password: ${kafkaPassword}
      bootstrapServers: example.com:9092
      config:
        auto.offset.reset: earliest
      security:
        protocol: SASL_PLAINTEXT
  stream-processor-sample-example:
    type: mongodbatlas:StreamProcessor
    properties:
      projectId: ${projectId}
      workspaceName: ${example.instanceName}
      processorName: sampleProcessorName
      pipeline: |
        [
          {\"$source\": {\"connectionName\": \"${["example-sample"].connectionName}\"}},
          {\"$emit\": {\"connectionName\": \"${["example-cluster"].connectionName}\", \"db\": \"sample\", \"coll\": \"solar\", \"timeseries\": {\"timeField\": \"_ts\"}}}
        ]
      state: STARTED
      tier: SP30
  stream-processor-cluster-to-kafka-example:
    type: mongodbatlas:StreamProcessor
    properties:
      projectId: ${projectId}
      workspaceName: ${example.instanceName}
      processorName: clusterProcessorName
      pipeline: |
        [
          {\"$source\": {\"connectionName\": \"${["example-cluster"].connectionName}\"}},
          {\"$emit\": {\"connectionName\": \"${["example-kafka"].connectionName}\", \"topic\": \"topic_from_cluster\"}}
        ]
      state: CREATED
  stream-processor-kafka-to-cluster-example:
    type: mongodbatlas:StreamProcessor
    properties:
      projectId: ${projectId}
      workspaceName: ${example.instanceName}
      processorName: kafkaProcessorName
      pipeline: |
        [
          {\"$source\": {\"connectionName\": \"${["example-kafka"].connectionName}\", \"topic\": \"topic_source\"}},
          {\"$emit\": {\"connectionName\": \"${["example-cluster"].connectionName}\", \"db\": \"kafka\", \"coll\": \"topic_source\", \"timeseries\": {\"timeField\": \"ts\"}}}
        ]
      state: CREATED
      tier: SP10
      options:
        dlq:
          coll: exampleColumn
          connectionName: ${["example-cluster"].connectionName}
          db: exampleDb
        autoscaling:
          minTier: SP10
          maxTier: SP50
variables:
  example-stream-processors:
    fn::invoke:
      function: mongodbatlas:getStreamProcessors
      arguments:
        projectId: ${projectId}
        workspaceName: ${example.instanceName}
  example-stream-processor:
    fn::invoke:
      function: mongodbatlas:getStreamProcessor
      arguments:
        projectId: ${projectId}
        workspaceName: ${example.instanceName}
        processorName: ${["stream-processor-sample-example"].processorName}
outputs:
  # example making use of functions
  streamProcessorsState: ${["example-stream-processor"].state}
  streamProcessorsResults: ${["example-stream-processors"].results}

terraform {
  required_providers {
    mongodbatlas = {
      source = "pulumi/mongodbatlas"
    }
  }
}

data "mongodbatlas_stream_processors" "example-stream-processors" {
  project_id     = projectId
  workspace_name = mongodbatlas_stream_instance.example.instance_name
}
data "mongodbatlas_stream_processor" "example-stream-processor" {
  project_id     = projectId
  workspace_name = mongodbatlas_stream_instance.example.instance_name
  processor_name = mongodbatlas_stream_processor.stream-processor-sample-example.processor_name
}

resource "mongodbatlas_stream_instance" "example" {
  lifecycle {
    create_before_destroy = true
  }
  project_id    = projectId
  instance_name = "InstanceName"
  data_process_region = {
    region         = "VIRGINIA_USA"
    cloud_provider = "AWS"
  }
}
resource "mongodbatlas_stream_connection" "example-sample" {
  lifecycle {
    create_before_destroy = true
  }
  project_id      = projectId
  workspace_name  = mongodbatlas_stream_instance.example.instance_name
  connection_name = "sample_stream_solar"
  type            = "Sample"
}
resource "mongodbatlas_stream_connection" "example-cluster" {
  lifecycle {
    create_before_destroy = true
  }
  project_id      = projectId
  workspace_name  = mongodbatlas_stream_instance.example.instance_name
  connection_name = "ClusterConnection"
  type            = "Cluster"
  cluster_name    = clusterName
  db_role_to_execute = {
    role = "atlasAdmin"
    type = "BUILT_IN"
  }
}
resource "mongodbatlas_stream_connection" "example-kafka" {
  lifecycle {
    create_before_destroy = true
  }
  project_id      = projectId
  workspace_name  = mongodbatlas_stream_instance.example.instance_name
  connection_name = "KafkaPlaintextConnection"
  type            = "Kafka"
  authentication = {
    mechanism = "PLAIN"
    username  = kafkaUsername
    password  = kafkaPassword
  }
  bootstrap_servers = "example.com:9092"
  config = {
    "auto.offset.reset" = "earliest"
  }
  security = {
    protocol = "SASL_PLAINTEXT"
  }
}
resource "mongodbatlas_stream_processor" "stream-processor-sample-example" {
  lifecycle {
    create_before_destroy = true
  }
  project_id     = projectId
  workspace_name = mongodbatlas_stream_instance.example.instance_name
  processor_name = "sampleProcessorName"
  pipeline       = <<-JSON
[
  {\"$source\": {\"connectionName\": \"${mongodbatlas_stream_connection.example-sample.connection_name}\"}},
  {\"$emit\": {\"connectionName\": \"${mongodbatlas_stream_connection.example-cluster.connection_name}\", \"db\": \"sample\", \"coll\": \"solar\", \"timeseries\": {\"timeField\": \"_ts\"}}}
]
JSON

  state          = "STARTED"
  tier           = "SP30"
}
resource "mongodbatlas_stream_processor" "stream-processor-cluster-to-kafka-example" {
  lifecycle {
    create_before_destroy = true
  }
  project_id     = projectId
  workspace_name = mongodbatlas_stream_instance.example.instance_name
  processor_name = "clusterProcessorName"
  pipeline       = <<-JSON
[
  {\"$source\": {\"connectionName\": \"${mongodbatlas_stream_connection.example-cluster.connection_name}\"}},
  {\"$emit\": {\"connectionName\": \"${mongodbatlas_stream_connection.example-kafka.connection_name}\", \"topic\": \"topic_from_cluster\"}}
]
JSON

  state          = "CREATED"
}
resource "mongodbatlas_stream_processor" "stream-processor-kafka-to-cluster-example" {
  lifecycle {
    create_before_destroy = true
  }
  project_id     = projectId
  workspace_name = mongodbatlas_stream_instance.example.instance_name
  processor_name = "kafkaProcessorName"
  pipeline       = <<-JSON
[
  {\"$source\": {\"connectionName\": \"${mongodbatlas_stream_connection.example-kafka.connection_name}\", \"topic\": \"topic_source\"}},
  {\"$emit\": {\"connectionName\": \"${mongodbatlas_stream_connection.example-cluster.connection_name}\", \"db\": \"kafka\", \"coll\": \"topic_source\", \"timeseries\": {\"timeField\": \"ts\"}}}
]
JSON

  state          = "CREATED"
  tier           = "SP10"
  options = {
    dlq = {
      coll            = "exampleColumn"
      connection_name = mongodbatlas_stream_connection.example-cluster.connection_name
      db              = "exampleDb"
    }
    autoscaling = {
      min_tier = "SP10"
      max_tier = "SP50"
    }
  }
}
# example making use of functions
output "streamProcessorsState" {
  value = data.mongodbatlas_stream_processor.example-stream-processor.state
}
output "streamProcessorsResults" {
  value = data.mongodbatlas_stream_processors.example-stream-processors.results
}

function getStreamProcessors(args: GetStreamProcessorsArgs, opts?: InvokeOptions): Promise<GetStreamProcessorsResult>
function getStreamProcessorsOutput(args: GetStreamProcessorsOutputArgs, opts?: InvokeOutputOptions): Output<GetStreamProcessorsResult>

def get_stream_processors(instance_name: Optional[str] = None,
                          project_id: Optional[str] = None,
                          workspace_name: Optional[str] = None,
                          opts: Optional[InvokeOptions] = None) -> GetStreamProcessorsResult
def get_stream_processors_output(instance_name: pulumi.Input[Optional[str]] = None,
                          project_id: pulumi.Input[Optional[str]] = None,
                          workspace_name: pulumi.Input[Optional[str]] = None,
                          opts: Optional[InvokeOutputOptions] = None) -> Output[GetStreamProcessorsResult]

func LookupStreamProcessors(ctx *Context, args *LookupStreamProcessorsArgs, opts ...InvokeOption) (*LookupStreamProcessorsResult, error)
func LookupStreamProcessorsOutput(ctx *Context, args *LookupStreamProcessorsOutputArgs, opts ...InvokeOption) LookupStreamProcessorsResultOutput

public static class GetStreamProcessors 
{
    public static Task<GetStreamProcessorsResult> InvokeAsync(GetStreamProcessorsArgs args, InvokeOptions? opts = null)
    public static Output<GetStreamProcessorsResult> Invoke(GetStreamProcessorsInvokeArgs args, InvokeOptions? opts = null)
    public static Output<GetStreamProcessorsResult> Invoke(GetStreamProcessorsInvokeArgs args, InvokeOutputOptions opts)
}

public static CompletableFuture<GetStreamProcessorsResult> getStreamProcessors(GetStreamProcessorsArgs args, InvokeOptions options)
public static Output<GetStreamProcessorsResult> getStreamProcessors(GetStreamProcessorsArgs args, InvokeOptions options)
public static Output<GetStreamProcessorsResult> getStreamProcessors(GetStreamProcessorsArgs args, InvokeOutputOptions options)

fn::invoke:
  function: mongodbatlas:index/getStreamProcessors:getStreamProcessors
  arguments:
    # arguments dictionary

data "mongodbatlas_get_stream_processors" "name" {
    # arguments
}