入力ストリームを使用してデータを送信する

mkdir input-stream-app-consume && cd input-stream-app-consume

import { HRN, OlpClientSettings, RequestFactory } from "@here/olp-sdk-core";
import { UserAuth, requestToken } from "@here/olp-sdk-authentication";
import { StreamApi } from "@here/olp-sdk-dataservice-api";

import { Kafka } from "kafkajs";
import { v4 as uuidv4 } from "uuid";

// Example telemetry data to be published.
const trackingId = "replace-with-your-tracking-id";
const streamIngestionData = {
  requestId: uuidv4(),
  type: "telemetry",
  version: "1.0",
  data: {
    id: trackingId,
    data: [
      {
        timestamp: Date.now(),
        position: {
          accuracy: 50,
          lat: 60.171318,
          lng: 24.941463
        }
      }
    ]
  }
};

// Access credentials and project information.
const accessKeyId = "replace-with-your-access-key-id";
const accessKeySecret = "replace-with-your-access-key-secret";

// Project HRN format: "hrn:here:authorization::<organization>:project/<projectId>"
const projectHrn = "replace-with-your-project-hrn";

// Catalog HRN format: "hrn:here:data::<organization>:tc-input-<projectId>"
const catalogHrn = "replace-with-your-catalog-hrn";
const layerId = "tc-ingest";

async function main(): Promise<void> {
  // Setup client settings.
  const userAuth = new UserAuth({
    tokenRequester: requestToken,
    credentials: { accessKeyId, accessKeySecret },
    scope: projectHrn
  });
  const olpClientSettings = new OlpClientSettings({
    environment: "here", getToken: () => userAuth.getToken()
  });
  // Use API lookup API to find a Stream v2 API endpoint.
  const requestBuilder = await RequestFactory.create(
    "stream", "v2", olpClientSettings, HRN.fromString(catalogHrn)
  );
  // Get kafka connection parameters from the Stream v2 API.
  const streamLayerEndpoint = await StreamApi.endpointByConsumer(
    requestBuilder, { layerId, type: "producer" }
  );

  // Create kafka producer.
  const kafka = new Kafka({
    clientId: streamLayerEndpoint.clientId,
    brokers: streamLayerEndpoint.bootstrapServers.map(s => `${s.hostname}:${s.port}`),
    ssl: true,
    sasl: {
      mechanism: "oauthbearer",
      oauthBearerProvider: async () => ({ value: await userAuth.getToken() })
    }
  });
  const producer = kafka.producer();

  // Publish telemetry.
  await producer.connect();
  await producer.send({
    topic: streamLayerEndpoint.topic,
    messages: [{
      key: trackingId, value: JSON.stringify(streamIngestionData)
    }]
  });
  await producer.disconnect();
}

main();

npm init --yes \
  && npm install --save-dev typescript \
  && npx tsc --init \
  && npm install --save-dev @types/node @types/uuid \
  && npm install --save \
    @here/olp-sdk-authentication \
    @here/olp-sdk-core \
    @here/olp-sdk-dataservice-api \
    kafkajs \
    uuid

npx tsx index.ts

# Use API lookup API to get base URL for Ingest v1 API.
catalogHrn="replace-with-your-catalog-hrn";
curl -X GET "https://api-lookup.data.api.platform.here.com/lookup/v1/resources/${catalogHrn}/apis/ingest/v1" \
  -H 'Authorization: Bearer '"${accessToken}"

# Post input stream telemetry object from telemetry.json file.
ingestV1BaseURL="replace-with-baseURL-from-api-lookup"
curl -X POST "${ingestV1BaseURL}/layers/tc-ingest" \
  -H 'Authorization: Bearer '"${accessToken}" \
  -H 'Content-Type: application/json' \
  -d @telemetry.json

{
  "error": {
    "title": "Bad Request",
    "status": 400,
    "code": "E902400",
    "cause": "Input stream ingestion validation failed",
    "action": "Correct the request and try again",
    "correlationId": "6f23e24a-aca2-4eee-861a-adf95e5902f2",
    "details": [
      {
        "name": "TrackingError",
        "correlationId": "6f23e24a-aca2-4eee-861a-adf95e5902f2",
        "code": 400,
        "reason": {
          "errors": [
            {
              "instancePath": "/data/data/0/position",
              "schemaPath": "#/properties/data/properties/data/items/properties/position/required",
              "keyword": "required",
              "params": {
                "missingProperty": "lng"
              },
              "message": "must have required property 'lng'"
            }
          ]
        }
      }
    ]
  },
  "type": "telemetry",
  "version": "1.0",
  "correlationId": "6f23e24a-aca2-4eee-861a-adf95e5902f2",
  "requestId": "6f23e24a-aca2-4eee-861a-adf95e5902f2"
}

import { HRN, OlpClientSettings } from "@here/olp-sdk-core";
import { UserAuth, requestToken } from "@here/olp-sdk-authentication";
import {
  PollRequest,
  StreamLayerClient,
  SubscribeRequest,
  UnsubscribeRequest,
} from "@here/olp-sdk-dataservice-read";

// Access credentials and project information.
const accessKeyId = "replace-with-your-access-key-id";
const accessKeySecret = "replace-with-your-access-key-secret";

// Project HRN format: "hrn:here:authorization::<organization>:project/<projectId>"
const projectHrn = "replace-with-your-project-hrn";

// Catalog HRN format: "hrn:here:data::<organization>:tc-input-<projectId>"
const catalogHrn = "replace-with-your-catalog-hrn";
const layerId = "tc-dead-letter";

async function main(): Promise<void> {
  // Setup client settings.
  const userAuth = new UserAuth({
    tokenRequester: requestToken,
    credentials: { accessKeyId, accessKeySecret },
    scope: projectHrn
  });
  const olpClientSettings = new OlpClientSettings({
    environment: "here", getToken: () => userAuth.getToken()
  });

  // Create StreamLayerClient.
  const streamLayerClient = new StreamLayerClient({
    catalogHrn: HRN.fromString(catalogHrn),
    layerId,
    settings: olpClientSettings,
  });

  // Create a subscription for the stream topic.
  const subscribeRequest = new SubscribeRequest().withMode("serial");
  const subscribtionId = await streamLayerClient.subscribe(subscribeRequest);

  // Consume data using poll method.
  const pollRequest = new PollRequest()
    .withMode("serial")
    .withSubscriptionId(subscribtionId);
  const messages = await streamLayerClient.poll(pollRequest);
  for (const message of messages) {
    // Base64 decode the message data.
    const data = JSON.parse(
      Buffer.from(message.metaData.data as string, 'base64').toString('utf-8')
    );
    console.log('Received message:', JSON.stringify(data, null, 2));
  }

  // Delete subscription from the stream topic.
  await streamLayerClient.unsubscribe(
    new UnsubscribeRequest()
      .withMode("serial")
      .withSubscriptionId(subscribtionId)
  );
}

main();