On this page

Show all

Help Center/ EventGrid/ API Reference/ APIs/ Event Stream Management/ Creating an Event Stream

Creating an Event Stream

Updated on 2025-05-06 GMT+08:00

Online Debugging

CLI Examples

View PDF

Function

This API is used to create an event stream.

URI

POST /v1/{project_id}/eventstreamings

**Table 1** Path Parameters
Parameter	Mandatory	Type	Description
project_id	Yes	String	Tenant resource space ID.

Request Parameters

**Table 2** Request header parameters
Parameter	Mandatory	Type	Description
X-Auth-Token	Yes	String	User token. The token can be obtained by calling the IAM API used to obtain a user token. The value of X-Subject-Token in the response header is the user token.

**Table 3** Request body parameters
Parameter	Mandatory	Type	Description
name	Yes	String	Event stream name. The value must be unique under a tenant. Only letters, digits, periods (.), underscores (_), and hyphens (-) are allowed. It must start with a letter or digit.
description	No	String	Description of the event stream.
source	Yes	EventStreamingSource object	Event source. An event stream has only one source.
sink	Yes	EventStreamingSink object	Event target. An event stream has only one event target. Either sink_fg or sink_kafka can be selected.
rule_config	No	rule_config object	Event rules, including filter and transform rules.
option	No	RunOption object	Running configuration.

**Table 4** EventStreamingSource
Parameter	Mandatory	Type	Description
source_kafka	No	SourceKafkaMQParameters object	Event source parameters.
source_mobile_rocketmq	No	SourceMobileMQParameters object	ECloud RocketMQ event source parameters
source_community_rocketmq	No	SourceCommunityMQParameters object	Open-Source RocketMQ event source parameters.
source_dms_rocketmq	No	SourceDMSMQParameters object	DMS event source parameters.
name	No	String	Event source type name.

**Table 5** SourceKafkaMQParameters
Parameter	Mandatory	Type	Description
addr	No	String	Kafka connection address.
group	Yes	String	Kafka consumer group.
instance_name	No	String	Kafka instance name.
security_protocol	No	String	Security protocol.
instance_id	No	String	Kafka instance ID.
topic	Yes	String	Kafka topic name.
seek_to	No	String	Consumption offset.
enable_sasl_ssl	No	Boolean	Whether to enable SASL_SSL.
sasl_mechanism	No	String	SASL authentication mechanism.
ssl_certificate_url	No	String	SASL certificate URL, which is a URL in OBS.
ssl_certificate_pwd	No	String	SASL certificate password.
user_name	No	String	Username.
password	No	String	Password.

**Table 6** SourceMobileMQParameters
Parameter	Mandatory	Type	Description
group_id	Yes	String	Consumer group ID.
instance_id	Yes	String	Instance ID.
topic	Yes	String	topic
tag	No	String	Tag.
authentication_required	No	Boolean	Indicates whether authentication is required.
msg_trace_switch	No	Boolean	Whether message tracing is required.
access_key	No	String	AccessKey
secret_key	No	String	SecretKey
message_model	Yes	String	Subscription mode.
addr_type	Yes	String	Access type.
addr	Yes	String	Address.
sdk_url	Yes	String	Client SDK.
consume_timeout	Yes	Integer	Consumption timeout.
message_type	Yes	String	Message type.
suspend_time	No	Integer	Retry waiting time.
max_reconsumer_times	No	Integer	Maximum retries.
consumer_thread_nums	No	Integer	Consumption threads.
consumer_batch_max_size	No	Integer	Maximum number of messages that can be consumed in a batch.
consumer_max_wait	No	Integer	Maximum waiting duration for batch consumption, in seconds.
vpc_id	No	String	VPC ID.
subnet_id	No	String	Subnet ID.

**Table 7** SourceCommunityMQParameters
Parameter	Mandatory	Type	Description
instance_name	No	String	Instance name. This parameter is required only for DMS for RocketMQ.
instance_id	No	String	Instance ID. This parameter is required only for DMS for RocketMQ.
addr	Yes	String	RocketMQ connection address.
group	Yes	String	Consumer group.
topic	Yes	String	Topic name.
tag	No	String	Tag.
vpc_id	Yes	String	VPC ID.
subnet_id	Yes	String	Subnet ID.
ssl_enable	No	Boolean	Whether to enable SSL.
enable_acl	No	Boolean	Indicates whether to enable ACL.
access_key	No	String	Username.
secret_key	No	String	Secret key.
message_type	No	String	Message type.
consume_timeout	No	Integer	Consumption timeout.
consumer_thread_nums	No	Integer	Number of consumption threads.
consumer_batch_max_size	No	Integer	Maximum number of messages that can be consumed in a batch.
max_reconsume_times	No	Integer	Maximum number of retries. The value -1 indicates that the retry is performed continuously.
suspend_current_queue_time_millis	No	Integer	Retry interval, in milliseconds.

**Table 8** SourceDMSMQParameters
Parameter	Mandatory	Type	Description
instance_name	No	String	Instance name. This parameter is required only for DMS for RocketMQ.
instance_id	Yes	String	Instance ID. This parameter is required only for DMS for RocketMQ.
group	Yes	String	Consumer group.
topic	Yes	String	Topic name.
tag	No	String	Tag.
ssl_enable	No	Boolean	Whether to enable SSL.
enable_acl	No	Boolean	Indicates whether to enable ACL.
access_key	No	String	Username.
secret_key	No	String	Secret key.
message_type	No	String	Consumption mode. The consumption mode varies according to the message production sequence. 1. Orderly production with orderly consumption: Messages are sent in order and consumed in the same order within a group. 2. Orderly production with concurrent consumption: Messages are sent in order and consumed concurrently, as close to order as possible. 3. Disorderly production with orderly consumption: Messages are sent out of order and consumed in queue order, not necessarily matching the send order. 4. Disorderly production with concurrent consumption: Messages are sent out of order and consumed concurrently, as close to order as possible.
engine_version	No	String	RocketMQ instance version.
consume_timeout	No	Integer	Consumption timeout.
consumer_thread_nums	No	Integer	Number of consumption threads.
consumer_batch_max_size	No	Integer	Maximum number of messages that can be consumed in a batch.
max_reconsume_times	No	Integer	Maximum number of retries. The value -1 indicates that the retry is performed continuously.
suspend_current_queue_time_millis	No	Integer	Retry interval, in milliseconds.

**Table 9** EventStreamingSink
Parameter	Mandatory	Type	Description
sink_fg	No	SinkFGParameters object	FunctionGraph event target parameters.
sink_kafka	No	SinkKafkaParameters object	Kafka event target parameters.
sink_obs	No	SinkObsParameters object	OBS event target parameters.
name	No	String	Event target name.

**Table 10** SinkFGParameters
Parameter	Mandatory	Type	Description
invoke_type	No	String	Function invocation mode, which can be asynchronous or synchronous.
urn	No	String	Function URN.
agency	No	String	Tenant agency.

**Table 11** SinkKafkaParameters
Parameter	Mandatory	Type	Description
topic	Yes	String	Topic name.
keyTransform	No	Array of TransForm objects	Key transform rule.
connectionId	Yes	String	Connection ID.

**Table 12** TransForm
Parameter	Mandatory	Type	Description
type	Yes	String	Type of the transform rule.
value	No	String	For a constant type rule, the field is the constant content definition. For a variable type rule, the field is a variable definition and the content must be a JSON object string. A maximum of 100 variables are supported, and nested structures are not supported. The variable name can contain a maximum of 64 characters. Only letters, digits, periods (.), underscores (_), and hyphens (-) are allowed. It must start with a letter or digit and cannot start with HC.. The variable value expression can be a constant or JSONPath expression. The string length cannot exceed 1024 characters.
template	No	String	Template definition of the rule content when the rule is of the variable type. Defined variables can be referenced.

**Table 13** SinkObsParameters
Parameter	Mandatory	Type	Description
access_key	Yes	String	AK
secret_key	Yes	String	SK
obs_bucket	Yes	String	Bucket.
obs_path	No	String	Dump directory.
time_format	Yes	String	Time directory format.

**Table 14** rule_config
Parameter	Mandatory	Type	Description
transform	No	TransForm object	Transform rule of the event subscription target.
filter	No	Object	Filter rule.

**Table 15** RunOption
Parameter	Mandatory	Type	Description
thread_num	No	Integer	Concurrency.
batch_window	No	BatchWindow object	Batch push.

**Table 16** BatchWindow
Parameter	Mandatory	Type	Description
count	No	Integer	Number of messages to push in a batch. Range: 1–10,000.
time	No	Integer	Number of retries.
interval	No	Integer	Batch push interval. Range: 0s to 15s.

Response Parameters

Status code: 200

**Table 17** Response header parameters
Parameter	Type	Description
X-Request-Id	String	This field is the request ID number for task tracking. Format is request_uuid-timestamp-hostname.

**Table 18** Response body parameters
Parameter	Type	Description
eventStreamingID	String	Event stream ID.

Example Requests

Create an event stream.

POST https://{eg_endpoint}/v1/{project_id}/eventstreamings

{
  "name" : "test-eventstreaming",
  "description" : "this is a test eventstreaming",
  "source" : {
    "source_kafka" : {
      "addr" : "10.10.10.10:8100",
      "group" : "group-test",
      "instance_name" : "instance-name-test",
      "security_protocol" : "PLAINTEXT",
      "instance_id" : "instance-id-test",
      "topic" : "topic-test",
      "seek_to" : "earliest",
      "enable_sasl_ssl" : true,
      "sasl_mechanism" : "PLAIN",
      "ssl_certificate_url" : "https://domain/kafka-certs.zip",
      "ssl_certificate_pwd" : null,
      "user_name" : "",
      "password" : ""
    },
    "source_mobile_rocketmq" : {
      "group_id" : "string",
      "instance_id" : "string",
      "topic" : "string",
      "tag" : "string",
      "authentication_required" : true,
      "msg_trace_switch" : true,
      "access_key" : "string",
      "secret_key" : "string",
      "message_model" : "CLUSTERING",
      "addr_type" : "PUBLIC",
      "addr" : "string",
      "sdk_url" : "string",
      "consume_timeout" : 30000,
      "message_type" : "string",
      "suspend_time" : 1800,
      "max_reconsumer_times" : 3,
      "consumer_thread_nums" : 20,
      "consumer_batch_max_size" : 20,
      "consumer_max_wait" : 5,
      "vpc_id" : "string",
      "subnet_id" : "string"
    },
    "source_community_rocketmq" : {
      "instance_name" : "instance-name-test",
      "instance_id" : "instance-id-test",
      "addr" : "10.10.10.10:8100",
      "group" : "group-test",
      "topic" : "topic-test",
      "tag" : "string",
      "vpc_id" : "string",
      "subnet_id" : "string",
      "ssl_enable" : true,
      "enable_acl" : true,
      "access_key" : "string",
      "secret_key" : "string",
      "message_type" : "string",
      "consume_timeout" : 30000,
      "consumer_thread_nums" : 20,
      "consumer_batch_max_size" : 1,
      "max_reconsume_times" : -1,
      "suspend_current_queue_time_millis" : 1000
    },
    "source_dms_rocketmq" : {
      "instance_name" : "instance-name-test",
      "instance_id" : "instance-id-test",
      "group" : "group-test",
      "topic" : "topic-test",
      "tag" : "string",
      "ssl_enable" : true,
      "enable_acl" : true,
      "access_key" : "string",
      "secret_key" : "string",
      "message_type" : "string",
      "engine_version" : "string",
      "consume_timeout" : 30000,
      "consumer_thread_nums" : 20,
      "consumer_batch_max_size" : 1,
      "max_reconsume_times" : -1,
      "suspend_current_queue_time_millis" : 1000
    },
    "name" : "string"
  },
  "sink" : {
    "sink_fg" : {
      "invoke_type" : "SYNC/ASYNC",
      "urn" : "String",
      "agency" : "string"
    },
    "sink_kafka" : {
      "topic" : "string",
      "keyTransform" : [ {
        "type" : "VARIABLE",
        "value" : "{\"contant_boolean\": true,\"contant_string\": \"constant\",\"varaible_string\": \"$.data.string\",\"varaible_json_object\": \"$.data.object\"}",
        "template" : "{\"contant_boolean\": ${contant_boolean},\"contant_string\": \"${contant_string!\\\"default\\\"}\",\"varaible_string\": \"${contant_boolean}\",\"varaible_json_object\": ${varaible_json_object!\"null\"}}"
      } ],
      "connectionId" : "string"
    },
    "sink_obs" : {
      "access_key" : "string",
      "secret_key" : "string",
      "obs_bucket" : "string",
      "obs_path" : "string",
      "time_format" : "string"
    },
    "name" : "string"
  },
  "rule_config" : {
    "transform" : {
      "type" : "VARIABLE",
      "value" : "{\"contant_boolean\": true,\"contant_string\": \"constant\",\"varaible_string\": \"$.data.string\",\"varaible_json_object\": \"$.data.object\"}",
      "template" : "{\"contant_boolean\": ${contant_boolean},\"contant_string\": \"${contant_string!\\\"default\\\"}\",\"varaible_string\": \"${contant_boolean}\",\"varaible_json_object\": ${varaible_json_object!\"null\"}}"
    },
    "filter" : { }
  },
  "option" : {
    "thread_num" : "3",
    "batch_window" : {
      "count" : 500,
      "time" : 10,
      "interval" : 0
    }
  }
}

Example Responses

Status code: 200

Event stream created.

{
  "eventStreamingID" : "23709d68-54d5-423b-a6be-03302e893152"
}

Status Codes


Status Code	Description
200	Event stream created.

Error Codes

See Error Codes.

Parent topic: Event Stream Management

Previous topic: Event Stream Management

Next topic: Obtaining the Event Stream List

Feedback

Was this page helpful?

Helpful Not helpful

Provide feedback

Thank you very much for your feedback. We will continue working to improve the documentation.See the reply and handling status in My Cloud VOC.

The system is busy. Please try again later.

Which of the following issues have you encountered?

Content is inconsistent with the product UI

Unclear descriptions

Lack of examples or code

Incorrect steps

Can't find what I need

Lack of best practices

Feedback (optional)

0/500

Select at least one type of issue, and enter your comments or suggestions.

Enter a maximum of 500 characters.

Submit Cancel

For any further questions, feel free to contact us through the chatbot.

Chatbot