How to use the @pulumi/aws.sns function in @pulumi/aws

//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

import * as pulumi from "@pulumi/pulumi";
import * as aws from "@pulumi/aws";
import * as fetch from "node-fetch";

// Examples of different types of metrics and alarms that can be set.

const topic = new aws.sns.Topic("sites-to-process-topic");
const subscription = topic.onEvent("for-each-url", async (event) => {
    const records = event.Records || [];
    for (const record of records) {
        const url = record.Sns.Message;

        console.log(`${url}: Processing`);

        // Fetch the contents at the URL
        console.log(`${url}: Getting`);
        try {
            const res = await fetch.default(url);
        } catch (err) {
            console.log(`${url}: Failed to GET`);
            return;
        }
    }

// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

import * as aws from "@pulumi/aws";
import * as serverless from "@pulumi/aws-serverless";

const topic = new aws.sns.Topic("sites-to-process-topic", { });

serverless.topic.subscribe("for-each-url", topic, async (event) => {
    const fetch = (await import("node-fetch")).default;

    const records = event.Records || [];
    for (const record of records) {
        const url = record.Sns.Message;

        console.log(`${url}: Processing`);

        // Fetch the contents at the URL
        console.log(`${url}: Getting`);
        try {
            const res = await fetch(url);
        } catch (err) {
            console.log(`${url}: Failed to GET`);

//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

import * as aws from "@pulumi/aws";
import * as awsx from "@pulumi/awsx";
import * as pulumi from "@pulumi/pulumi";
import * as fetch from "node-fetch";

// Examples of different types of metrics and alarms that can be set.

const topic = new aws.sns.Topic("sites-to-process-topic");
const subscription = topic.onEvent("for-each-url", async (event) => {
    const records = event.Records || [];
    for (const record of records) {
        const url = record.Sns.Message;

        console.log(`${url}: Processing`);

        // Fetch the contents at the URL
        console.log(`${url}: Getting`);
        try {
            const res = await fetch.default(url);
        } catch (err) {
            console.log(`${url}: Failed to GET`);
            return;
        }
    }

//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

import * as aws from "@pulumi/aws";
import * as awsx from "@pulumi/awsx";
import * as pulumi from "@pulumi/pulumi";
import * as fetch from "node-fetch";

// Examples of different types of metrics and alarms that can be set.

const topic = new aws.sns.Topic("sites-to-process-topic");
const subscription = topic.onEvent("for-each-url", async (event) => {
    const records = event.Records || [];
    for (const record of records) {
        const url = record.Sns.Message;

        console.log(`${url}: Processing`);

        // Fetch the contents at the URL
        console.log(`${url}: Getting`);
        try {
            const res = await fetch.default(url);
        } catch (err) {
            console.log(`${url}: Failed to GET`);
            return;
        }
    }

//     users: [user],
//     groups: [group],
//     roles: [role],
//     policyArn: policy.arn,
// });

// Kinesis
const stream = new aws.kinesis.Stream("mystream", {
    shardCount: 1,
});

// SQS
const queue = new aws.sqs.Queue("myqueue");

// SNS
const topic = new aws.sns.Topic("mytopic");

const topicSubscription = new aws.sns.TopicSubscription("mytopicsubscription", {
    topic: topic,
    protocol: "sqs",
    endpoint: queue.arn,
});

constructor(name: string, opts?: pulumi.ResourceOptions) {
        super("cloud:topic:Topic", name, {}, opts);

        this.name = name;
        this.topic = new aws.sns.Topic(name, {}, { parent: this });
        this.subscriptions = [];
        const topicId = this.topic.id;

        this.publish = async (item) => {
            const awssdk = await import("aws-sdk");
            const snsconn = new awssdk.SNS();
            const result = await snsconn.publish({
                Message: JSON.stringify(item),
                TopicArn: topicId.get(),
            }).promise();
        };

        this.registerOutputs({
            topic: this.topic,
        });
    }

export function getUnhandledErrorTopic(): aws.sns.Topic {
    if (!unhandledErrorTopic) {
        unhandledErrorTopic = new aws.sns.Topic(
            createNameWithStackInfo(`unhandled-error`),
            undefined,
            { parent: getGlobalInfrastructureResource() });
    }

    return unhandledErrorTopic;
}

export function enableAlarm(instance: aws.ec2.Instance, threshold: number): void {
    if (emailAddress === undefined) {
        throw new Error("Missing email address configuration");
    }

    let topic = new aws.sns.Topic(instance.urnName + "-topic", {});
    let topicSubscription = new aws.sns.TopicSubscription(instance.urnName + "-subscription", {
        topic: topic,
        protocol: "email",
        endpoint: emailAddress,
    });
    let alarm = new aws.cloudwatch.MetricAlarm(instance.urnName + "-alarm", {
        alarmActions: [ topic ],
        metricName: "CPUUtilization",
        namespace: "AWS/EC2",
        statistic: "Average",
        period: 60,
        evaluationPeriods: 3,
        threshold: threshold,
        comparisonOperator: "GreaterThanThreshold",
        dimensions: [{
            name: "InstanceId",
            value: instance,

resName,
        (ev: SNSEvent, ctx: aws.serverless.Context, cb: (error: any, result: any) => void) => {
            Promise.all(ev.Records.map(async (record: SNSRecord) => {
                await handler(record.Sns);
            }))
            .then(() => { cb(null, null); })
            .catch((err: any) => { cb(err, null); });
        },
    );
    const invokePermission = new aws.lambda.Permission(resName, {
        action: "lambda:invokeFunction",
        function: func.lambda,
        principal: "sns.amazonaws.com",
        sourceArn: topic.id,
    });
    subscription = new aws.sns.TopicSubscription(resName, {
        topic: topic,
        protocol: "lambda",
        endpoint: func.lambda.arn,
    });

    return subscription;
}

export function enableAlarm(instance: aws.ec2.Instance, threshold: number): void {
    if (emailAddress === undefined) {
        throw new Error("Missing email address configuration");
    }

    let topic = new aws.sns.Topic(instance.urnName + "-topic", {});
    let topicSubscription = new aws.sns.TopicSubscription(instance.urnName + "-subscription", {
        topic: topic,
        protocol: "email",
        endpoint: emailAddress,
    });
    let alarm = new aws.cloudwatch.MetricAlarm(instance.urnName + "-alarm", {
        alarmActions: [ topic ],
        metricName: "CPUUtilization",
        namespace: "AWS/EC2",
        statistic: "Average",
        period: 60,
        evaluationPeriods: 3,
        threshold: threshold,
        comparisonOperator: "GreaterThanThreshold",
        dimensions: [{
            name: "InstanceId",