出力ストリームからトレースを受け取る

mkdir traces-stream-app-poll && cd traces-stream-app-poll

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";
import util from 'util';

// 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-output-<projectId>"
const catalogHrn = "replace-with-your-catalog-hrn";

const layerId = "tc-traces";

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 subscriptionId = await streamLayerClient.subscribe(subscribeRequest);

  // Consume data using poll method.
  const pollRequest = new PollRequest()
    .withMode("serial")
    .withSubscriptionId(subscriptionId);

  let interrupted = false;
  process.on('SIGINT', () => { interrupted = true; });

  console.log('Consuming traces (press Ctrl+C to exit)...');

  try {
    while (!interrupted) {
      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(util.inspect(data, { depth: null, colors: true }));
      }
    }
  } finally {
    // Delete subscription from the stream topic.
    await streamLayerClient.unsubscribe(
      new UnsubscribeRequest()
        .withMode("serial")
        .withSubscriptionId(subscriptionId)
    );
  }
}

main();

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

npx tsc && node .

mkdir traces-stream-app-consume && cd traces-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, logLevel } from "kafkajs";
import util from 'util';

// 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-output-<projectId>"
const catalogHrn = "replace-with-your-catalog-hrn";

const layerId = "tc-traces";

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: "consumer" }
  );

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

  // Subscribe and listen for traces.
  await consumer.connect();
  await consumer.subscribe({ topic: streamLayerEndpoint.topic });
  await consumer.run({
    eachMessage: async ({ message }) => {
      const data = JSON.parse((message.value || "{}").toString());
      console.log(util.inspect(data, { depth: null, colors: true }));
    },
  });

  // Disconnect and exit on Ctrl+C.
  console.log('Consuming traces (press Ctrl+C to exit)...');
  process.on('SIGINT', async () => {
    console.log('Disconnecting...');
    await consumer.disconnect();
  });
}

main();

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

npx tsc && node .

timestamp=$(date +%s)
curl https://tracking.hereapi.com/v3/ \
  --request POST \
  --header 'authorization: Bearer {accessToken}' \
  --header 'content-type: application/json' \
  --data '{
  "id": "{trackingId}",
  "data": [
    {
      "timestamp": '$timestamp'000,
      "position": {
        "lat": "52.4988217",
        "lng": "13.35170005",
        "alt": "86",
        "accuracy": "25"
      }
    }
  ]
}'