将 Eclipse Ditto 连接到 Apache Kafka [英] Connecting Eclipse Ditto to Apache Kafka

问题描述

我正在按照本文档中解释的说明将 Apache Kafka 连接到 Eclipse Ditto.

I am following the instructions explained in this documentation to connect Apache Kafka to Eclipse Ditto.

https://www.eclipse.org/ditto/connectivity-protocol-bindings-kafka2.html

我不确定以下内容.

1) ["ditto:outbound-auth-subject", "..."] 在授权上下文中.

1) ["ditto:outbound-auth-subject", "..."] under the Authorization context.

2) "address": "topic/key"

2) "address": "topic/key"

请让我知道他们！提前致谢.

Please let me know about them! Thank you in advance.!

请找出我用来连接 Ditto 和 Kafka 的命令

Please find the command that I used to connect Ditto and Kafka

curl -X POST -i -u devops:foobar -H 'Content-Type: application/json' -d '{
"targetActorSelection": "/system/sharding/connection",
"headers": {
    "aggregate": false
},
"piggybackCommand": {
    "type": "connectivity.commands:createConnection",
    "connection": {
        "id": "MyKafkaConnection1",
        "connectionType": "kafka",
        "connectionStatus": "open",
        "uri": "tcp://radsah:password@localhost:9092",
        "specificConfig": {
        "bootstrapServers": "10.196.2.218:9092",
        "saslMechanism": "plain"
        },
        "failoverEnabled": true,
        "targets": [
              {
             "address": "digital-twins",
              "topics": [
               "_/_/things/twin/events",
               "_/_/things/live/messages"
                            ],
            "authorizationContext": ["ditto:outbound-auth-subject"]
        }],
        "mappingContext": {
            "mappingEngine": "JavaScript",
            "options": {
                "incomingScript": "function mapToDittoProtocolMsg(\n    headers,\n    textPayload,\n    bytePayload,\n    contentType\n) {\n\n    if (contentType !== \"application/json\") {\n        return null;\n    }\n\n    var jsonData = JSON.parse(textPayload);\n    var temperature = jsonData.temp;\n    var humidity = jsonData.hum;\n    \n    var path;\n    var value;\n    if (temperature != null && humidity != null) {\n        path = \"/features\";\n        value = {\n                temperature: {\n                    properties: {\n                        value: temperature\n                    }\n                },\n                humidity: {\n                    properties: {\n                        value: humidity\n                    }\n                }\n            };\n    } else if (temperature != null) {\n        path = \"/features/temperature/properties/value\";\n        value = temperature;\n    } else if (humidity != null) {\n        path = \"/features/humidity/properties/value\";\n        value = humidity;\n    }\n    \n    if (!path || !value) {\n        return null;\n    }\n\n    return Ditto.buildDittoProtocolMsg(\n        \"org.eclipse.ditto\",\n        headers[\"device_id\"],\n        \"things\",\n        \"twin\",\n        \"commands\",\n        \"modify\",\n        path,\n        headers,\n        value\n    );\n}"
            }
        }
    }
}
}' http://localhost:8080/devops/piggyback/connectivity?timeout=8000

我已经使用 Hono 注册了一个设备，我正在将数据发送到 Ditto.同上成功接收数据.但我想将收到的数据发送给 Kafka.

I have registered a device using Hono and I am sending the data to Ditto. Ditto successfully receives the data. But I want send this received data to Kafka.

Kafka 和 Ditto 连接成功.但我没有收到卡夫卡消费者的数字双胞胎".我错过了什么吗?

Connection is successfully established between Kafka and Ditto. But I am not receiving at the kafka-consumer "digital-twins". Am I missing something?

使用策略命令

curl -X PUT 'http://localhost:8080/api/2/policies/org.eclipse.ditto:5100' -u 'ditto:ditto' -H 'Content-Type: application/json' -d '{
"entries": {
    "owner": {
        "subjects": {
            "nginx:ditto": {
                "type": "nginx basic auth user"
            }
        },
        "resources": {
            "thing:/": {
                "grant": [
                    "READ","WRITE"
                ],
                "revoke": []
            },
            "policy:/": {
                "grant": [
                    "READ","WRITE"
                ],
                "revoke": []
            },
            "message:/": {
                "grant": [
                    "READ","WRITE"
                ],
                "revoke": []
            }
        }
    }
}
}

推荐答案

关于授权上下文你可以看看 我们连接文档中的授权部分.它必须包含在您的事物的策略或 ACL 中定义的主题.

regarding the authorization context you can have a look at the authorization section in our connections documentation. It has to hold a subject that is defined in the policy or ACL of your things.

例如:

事物foo:bar"的策略已经定义了主题somePrefix:someValue"的整个事物的读取访问权限.

The policy of Thing "foo:bar" has read access of the whole thing for subject "somePrefix:someValue" defined.

{
  "policyId": "foo:bar",
  "entries": {
    ... //Maybe more entries
    "MyKafkaConnection": {
      "subjects": {
        "somePrefix:someValue": {
          "type": "my description for this subject"
        }
      },
      "resources": {
        "thing:/": {
          "grant": [
            "READ"
          ],
          "revoke": []
        },
        "message:/": {
          "grant": [
            "READ"
          ],
          "revoke": []
        }
      }
    }
  }
}

在您所指的示例中，与foo:bar"相关的事件将通过 kafka 连接发布到您在地址字段中指定的主题上.

In the sample you referring to, the events related to "foo:bar" would then be published via the kafka connection on the topic you specified in the address field.

这篇关于将 Eclipse Ditto 连接到 Apache Kafka的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！