Add a way to send events to the Event Hub

azure-ts-stream-analytics/README.md

@@ -56,3 +56,11 @@ An example Pulumi program that deploys an Azure Stream Analytics job to transfor
  [{"Make":"Kia","Sales":3};{"Make":"Honda","Sales":1}]
 
  ```
+
+ You can send a message with a `curl` command:
+
+ ```
+ curl -X POST '$(pulumi stack output inputEndpoint)' -H 'Authorization: $(pulumi stack output sasToken)' -H 'Content-Type: application/atom+xml;type=entry;charset=utf-8' -d '{"Make":"Kia","Sales":2,"Time":"2019-06-26T10:22:36Z"}'
+ ```
+
+1. [Start the Stream Analytics job](https://docs.microsoft.com/en-us/azure/stream-analytics/start-job). The job will start emitting messages to the output Event Hub once per minute. The Azure Function `analytics-output` will start printing those events into the console (you'd have to open the function console in the Azure portal to see them).

azure-ts-stream-analytics/index.ts

@@ -1,6 +1,8 @@
 // Copyright 2016-2019, Pulumi Corporation. All rights reserved.
 
+import * as pulumi from "@pulumi/pulumi";
 import * as azure from "@pulumi/azure";
+import { createSharedAccessToken } from "./token";
 
 // Create an Azure Resource Group
 const resourceGroup = new azure.core.ResourceGroup("streams-rg");
@@ -93,3 +95,7 @@ outputHub.onEvent("analytics-output", {
  console.log(JSON.stringify(event));
  },
 });
+
+const url = pulumi.interpolate`https://${namespace.name}.servicebus.windows.net`;
+export const sasToken = pulumi.all([url, namespace.defaultPrimaryKey]).apply(([u, pk]) => createSharedAccessToken(u, "RootManageSharedAccessKey", pk));
+export const inputEndpoint = pulumi.interpolate`${url}/${inputHub.name}/messages?timeout=60&api-version=2014-01`;

azure-ts-stream-analytics/package.json

@@ -2,10 +2,13 @@
  "name": "azure-ts-stream-analytics",
  "version": "0.1.0",
  "devDependencies": {
- "@types/node": "^10.3.3"
+ "@types/node": "^10.3.3",
+ "@types/utf8": "^2.1.6"
  },
  "dependencies": {
  "@pulumi/azure": "latest",
- "@pulumi/pulumi": "latest"
+ "@pulumi/pulumi": "latest",
+ "crypto": "^1.0.1",
+ "utf8": "^3.0.0"
  }
-}
+}

azure-ts-stream-analytics/token.ts

@@ -0,0 +1,16 @@
+import { encode } from "utf8";
+import { createHmac } from "crypto";
+
+export function createSharedAccessToken(uri: string, saName: string, saKey: string): string {
+ if (!uri || !saName || !saKey) {
+ throw "Missing required parameter";
+ }
+ var encoded = encodeURIComponent(uri);
+ var now = new Date();
+ var week = 60 * 60 * 24 * 7;
+ var ttl = Math.round(now.getTime() / 1000) + week;
+ var signature = encoded + '\n' + ttl;
+ var signatureUTF8 = encode(signature);
+ var hash = createHmac('sha256', saKey).update(signatureUTF8).digest('base64');
+ return 'SharedAccessSignature sr=' + encoded + '&sig=' + encodeURIComponent(hash) + '&se=' + ttl + '&skn=' + saName;
+}

azure-ts-stream-analytics/tsconfig.json

@@ -17,6 +17,7 @@
  "strictNullChecks": true
  },
  "files": [
- "index.ts"
+ "index.ts",
+ "token.ts"
  ]
-}
+}