This document introduces how to connect to MatrixGate using API interfaces to achieve high - speed data ingestion.
MatrixGate provides an HTTP API that supports various programming languages to import data into the YMatrix database through HTTP interfaces. The MatrixGate HTTP protocol format and response codes are as follows:
MatrixGate HTTP Protocol Format
| Protocol Type | Protocol Format | Usage and Example |
|---|---|---|
| URL | http://\ |
Specifies the mxgate connection URL |
| PATH | / | Currently supports /, ignoring any PATH after / |
| HTTP Method | POST | Currently supports loading data via POST method |
| HTTP Header | Content-Encoding: gzip | Currently supports gzip compression for HTTP Body content |
| Content-Type: text/plain | Currently supports text/plain | |
| HTTP Body | SchemaName.TableName Timestamp|ID|C1|C2|..|Cn |
The first line of the Body format specifies the target table for data loading. SchemaName can be omitted (default is Public), while TableName is mandatory. The second line and onwards represent time - series data rows, with each row corresponding to a row in the target table. Columns are separated by |, and rows are separated by \n. The first field of each row is a timestamp in UNIX format (seconds), as described in --time-format. The second field is TagID, an integer. The third to the last fields correspond to the columns of the target table. It is recommended that the DDL definition of the target table also follows the column order of (Timestamp, TagID, C1, C2, …, Cn) |
MatrixGate HTTP Response Codes
| Response Code | Meaning | Notes |
|---|---|---|
| 200 | StatusOK | Partial data format errors. The response Body will contain error lines with error messages, such as:At line: 2missing data for column "c3" |
| 204 | StatusNoContent | Data successfully loaded into MatrixGate |
| 400 | StatusBadRequest | Data request error, such as POST BODY format error, target table not found, data compression format mismatch with HTTP request header, etc. |
| 405 | StatusMethodNotAllowed | Non - POST HTTP request |
| 408 | StatusTimeout | Request timeout |
| 500 | StatusIntervalServerError | Database - side error, data loading failed. The response Body contains detailed error information |
| 503 | StatusServiceUnavailable | MatrixGate rejects the request, such as exceeding the maximum number of connections, or MatrixGate is shutting down, etc. |
First, create the table testtable in the demo database.
=# CREATE TABLE testtable (
time TIMESTAMP WITH TIME ZONE,
tagid INT,
c1 INT,
c2 INT,
c3 INT
)USING MARS3
DISTRIBUTED BY (tagid)
ORDER BY (time,tagid);Generate the mxgate.conf configuration file.
$ mxgate config --db-database testdb \
--db-master-host localhost \
--db-master-port 5432 \
--db-user mxadmin \
--db-password 123123 \
--target public.testtable \
--format csv \
--time-format unix-second \
--delimiter '|' \
--parallel 256 \
--stream-prepared 3 \
--interval 250 \
--transform plain \
> mxgate.confEdit the data loading file data.txt.
$ vi data.txt
public.testtable
1603777821|1|101|201|301
1603777822|2|102|202|302
1603777823|3|103|203|303Start mxgate with the generated configuration file mxgate.conf.
$ mxgate --config mxgate.confSend the HTTP request to load the data.
$ curl http://localhost:8086/ -X POST -H 'Content-Type: text/plain' --data-binary "@data.txt"Connect to the database to check if the data has been loaded successfully.
$ psql demo
demo=# SELECT extract(epoch FROM "time"), * FROM testtable;
date_part | time | tagid | c1 | c2 | c3
------------+------------------------+-------+-----+-----+-----
1603777821 | 2020-10-27 13:50:21+08 | 1 | 101 | 201 | 301
1603777822 | 2020-10-27 13:50:22+08 | 2 | 102 | 202 | 302
1603777823 | 2020-10-27 13:50:23+08 | 3 | 103 | 203 | 303
(3 rows)Add the following two key - value pairs to the Header of each HTTP POST request sent to mxgate:
"Batch-Type" = json
"Job-Name" = public.t1"Batch-Type" = json indicates that the data format in the POST request Body is JSON.Job-Name specifies the corresponding table name for data insertion, in the format: schemaname.tablename.Add the following transform - related configuration information to the mxgate configuration file:
[transform]
## Overall parallel level for transform, only for non - strict mode
# parallel = 256
## Transform decodes input data and performs type/format conversion
## Types restricted to: plain/json/nil/mxmon/hanangdbc
## transform = "plain"
transform = "json"
[transform.json]
mapping = [{
table-name = "public.t1",
field-map = [{dest="id", source="$.id", enabled=true}, {dest="content", source="$.content", enable=false}]
}]source = "$.id" is a Json Path expression. You can refer to JsonPath.Assuming the target table structure is as follows:
Table "public.t1"
Column | Type | Collation | Nullable | Default
---------+---------+-----------+----------+---------
id | integer | | |
content | text | | |
Distributed by: (id)The JSON data sample structure is:
{"id": 12345, "content": "this is a sample json"}The JSON transform feature of mxgate will extract the values of id and content from the corresponding JSON data according to the [transform.json] configuration, and then concatenate them into CSV or text format to load into the database.
The result is as follows:
=# SELECT * FROM t1;
id | content
-------+-----------------------
12345 | this is a simple json
(1 row)Currently, when writing JSON via HTTP, mxgate only supports the format where each line is a complete JSON string. If there is a line break in the middle, the write operation will fail.
For example, the following JSON data can be transmitted:
{"employees": [{"id": 1, "name": "John Doe", "position": "Manager", "department": "Sales"},{"id": 2,"name": "Jane Smith","position": "Engineer","department": "Engineering"}]}The following will fail:
{
"employees": [
{
"id": 1,
"name": "John Doe",
"position": "Manager",
"department": "Sales"
},
{
"id": 2,
"name": "Jane Smith",
"position": "Engineer",
"department": "Engineering"
}
]
}An SDK (Software Development Kit) is a set of tools that can free developers from non - business - logic feature development, significantly improving development efficiency and experience, and enhancing usability.
You can introduce the SDK JAR package in the following ways:
Note!
You only need to choose one of the above methods. It is recommended to use (1) or (2) to directly introduce the SDK from the Maven or Gradle remote repository, which is efficient and convenient. For the local import method (3), refer to MatrixGate FAQ: 21. It is not commonly used and will not be detailed here.
(1) Fetch SDK Package Automatically from Maven Remote Repository
Configure the following dependency in your project's pom.xml file:
<dependencies>
<dependency>
<groupId>cn.ymatrix</groupId>
<artifactId>mxgate-sdk-java</artifactId>
<version>1.1.2</version>
</dependency>
</dependencies>(2) Introduce SDK Dependency via Gradle Remote Repository
repositories {
mavenCentral()
}
dependencies {
implementation 'cn.ymatrix:mxgate-sdk-java:1.0.20'
}YMatrix supports writing data via gRPC and HTTP. Specific instructions are as follows:
Note!
A YMatrix version that supports gRPC is required, i.e., version 4.6.1 and later.
Create the mxgate_grpc.conf configuration file on the Master node and start mxgate.
Note!
Before using mxgate, you need to create the table in the database first. In the example, the Master host is mdw, the database isdemo, and the data table istest_table_transfer.
The code is as follows:
# Create mxgate config file
[mxadmin@mdw ~]$ mxgate config \
--source grpc \
--db-database demo \
--target public.test_table_transfer \
--time-format raw \
--grpc-port 8087 \
--format csv \
> mxgate_grpc.conf
# Start mxgate
[mxadmin@mdw ~]$ mxgate start --config mxgate_grpc.confAs shown in the example code, when using the SDK to write data to mxgate, you need to specify the source parameter of mxgate as grpc and the format as csv. Since the SDK needs to know where to write the data, you also need to specify the gRPC port number in the configuration file, which is 8087 in the example.
Note!
Even when starting mxgate via HTTP, you still need to specify the gRPC port number of mxgate. This is because the SDK still uses gRPC to obtain database table metadata from mxgate, although the data writing method is switched to HTTP.
Create the mxgate_http.conf configuration file on the Master node and start mxgate. In the example, the Master hostname is mdw. The code is as follows:
# Create mxgate config file
[mxadmin@mdw ~]$ mxgate config \
--source http \
--db-database demo \
--target public.test_table_transfer \
--time-format raw \
--grpc-port 8087 \
--format csv \
> mxgate_http.conf
# Start mxgate
[mxadmin@mdw ~]$ mxgate start --config mxgate_http.confThis section will introduce both asynchronous and synchronous sending methods. You can choose according to your needs.
The example table structure is as follows:
Partitioned table "public.test_table_transfer"
Column | Type | Collation | Nullable | Default
--------+-----------------------------+-----------+----------+---------
ts | timestamp without time zone | | |
tag | integer | | not null |
c1 | double precision | | |
c2 | double precision | | |
c3 | double precision | | |
c4 | double precision | | |
c5 | text | | |
c6 | text | | |
c7 | text | | |
c8 | text | | |
Partition key: RANGE (ts)First, perform global initialization settings:
// Set log level, default is INFO.
MxLogger.loggerLevel(LoggerLevel.INFO);
// Logs will be output to stdout by default. If you do not want to output, you can pass false in the following API.
MxLogger.enableStdout(true);
// The default log file path and naming format of the SDK is /tmp/mxgate_sdk_java_2022-08-26_133403.log,
// Users can customize the output log file path and file name.
MxLogger.writeToFile("/tmp/mxgate_sdk_java.log");Asynchronous sending appends the Tuple to an internal queue of the SDK, and then sends the data via asynchronous HTTP requests. First, you need to start the MxBuilder, which is a singleton and globally unique.
Note!
You only need to build once in a project. You can perform related configurations before initializing the MxBuilder.
MxBuilder builder = MxBuilder.newBuilder()
.withDropAll(false) // If used for testing, set to true to drop data instead of sending to mxgate; set to false to send data to mxgate
.withCacheCapacity(100000) // Size of the queue for temporarily storing tuples in batches
.withCacheEnqueueTimeout(2000) // Timeout for tuples to be enqueued if the queue is full. Throws IllegalStateException if exceeded
.withConcurrency(10) // Number of threads writing data to mxgate concurrently
.withRequestTimeoutMillis(3000) // Timeout for each data write request (in milliseconds)
.withMaxRetryAttempts(3) // Number of retries for each write request if an issue occurs
.withRetryWaitDurationMillis(3000) // Interval between retries (currently fixed)
.withRequestType(RequestType.WithHTTP) // SDK supports sending data to mxgate via HTTP and gRPC. Corresponding configurations are: RequestType.WithHTTP, RequestType.WithGRPC
.withCircuitBreaker() // Use the built - in circuit breaker. If the failure rate or slow request rate reaches the threshold, it will be activated. After activation, data sending to mxgate will be paused for 30 seconds, and tuples cannot be appended
.withMinimumNumberOfCalls(1) // Minimum number of calls for the circuit breaker to take effect (must be >= 1, default is 10)
.withSlidingWindowSize(10) // Size of the sliding window for calculating the failure rate (must be >= 1, default is 100)
.withFailureRateThreshold(60.0f) // Failure rate threshold (must be >0 and <= 100). If the failure rate reaches this threshold, the circuit breaker will be activated
.withSlowCallDurationThresholdMillis(1000) // Threshold for slow request duration (milliseconds). Requests exceeding this duration are considered slow (note that this duration should be less than the request timeout)
.withSlowCallRateThreshold(80.0f) // Slow request rate threshold. If the slow request rate reaches this threshold, the circuit breaker will be activated
// .withRequestAsync(true) // Uncomment this line to enable asynchronous mode for sending data to mxgate (optional)
// .withConcurrency(20) // Typically, asynchronous mode only requires tens of concurrency to achieve the same or even higher throughput as synchronous mode (optional)
// MxBuilder's builder pattern has added the following API to adjust the concurrency of CSV parallel transformation (at the MxClient Group level)
// .withCSVConstructionParallel(100) // (Supported from v1.1.0, uncomment this line to use)
.build();
// MxBuilder has added a singleton API. After MxBuilder is successfully built, you can obtain the globally unique singleton instance of MxBuilder at any location using the following API (supported from v1.1.2)
// MxBuilder.instance(); // (Supported from v1.1.2, uncomment this line to use)
// builder.getTupleCacheSize(); // Uncomment this line to get the number of remaining Tuples in the SDK's internal Tuple cache in real - time (optional)The connect method of the Builder takes four parameters:
(1) The hostname (IP address) and port number of the mxgate process providing the specific service for receiving data and database table metadata. For example, for HTTP, it is http://localhost:8086/; for gRPC, it is localhost:8087.
(2) Schema.
(3) Table name.
(4) Callback. If the Builder successfully connects to mxgate, the onSuccess method will be called back, returning an MxClient instance for data writing. If the connection fails, the onFailure method will be called back, and failureMsg will explain the reason for the failure.
The specific code is as follows. Remember to replace the port number, schema name, and table name with the actual content.
// Asynchronous method (supported from v1.0.13)
builder.connect("http://localhost:8086/", "localhost:8087", "public", "test_table", new ConnectionListener() {
@Override
public void onSuccess(MxClient client) {
sendData(client);
}
@Override
public void onFailure(String failureMsg) {
// Handle failure
}
});Through the onSuccess callback function of the ConnectionListener, you can obtain an MxClient instance. Subsequently, you can use the APIs provided by MxClient to write data to mxgate.
// Synchronous method (supported from v1.0.13)
MxClient client = builder.connect(httpHost, gRPCHost, schema, table);/*
* v1.1.0 has enhanced the scalability of MxClient. MxBuilder has added several new APIs.
* The MxClient obtained through these APIs belongs to a specific MxClient Group.
* You can freely define parameters such as the MxClient Group Number, Tuples cache capacity, and concurrency of TuplesConsumer using these APIs.
* (Supported from v1.1.0)
*/
// MxClient client = builder.connectWithGroup(dataSendingHost, metadataHost, schema, table, 10);
// MxClient client = builder.connectWithGroup(dataSendingHost, metadataHost, schema, table, 1, 1000, 3000, 10); // (Supported from v1.1.0, uncomment this line to use)
/*
* APIs starting with "skip" will skip the process of connecting to the backend service to obtain database table metadata.
* Such MxClient can only obtain a lightweight Tuple through generateEmptyTupleLite().
* (Supported from v1.1.0)
*/
// MxClient client = mxBuilder.skipConnectWithGroup(dataSendingHost, metadataHost, schema, table, delimiter, 1);
// MxClient client = mxBuilder.skipConnectWithGroup(dataSendingHost, metadataHost, schema, table, delimiter, 1, 1000, 3000, 1);Note!
MxClient is thread - safe. However, for optimal performance in multi - threaded concurrent use, it is best to have a separate MxClient in each thread, i.e., obtain multiple MxClient instances through multiple calls to MxBuilder's connect method.
private void sendData(MxClient client) {
if (client != null) {
/*
* MxClient accumulates a batch of Tuples and sends them to mxgate as a micro - batch.
* This API sets the waiting time for each micro - batch accumulation. The default is 2000 milliseconds,
* meaning that every 2 seconds, it will attempt to send a batch of data to mxgate. If no data is written within this 2 - second period,
* no data will be sent.
*/
client.withIntervalToFlushMillis(5000);
/*
* Sets the number of Tuple bytes to accumulate as a micro - batch for sending. Even if the flush interval time has not been reached,
* if the accumulated micro - batch byte count reaches this threshold, it will be sent. If the flush interval time is reached but the
* accumulated byte count has not reached the set threshold, it will still be sent.
*/
client.withEnoughBytesToFlush(500);
/*
* Compared to withEnoughBytesToFlush, this API improves performance when appending Tuples because it avoids calculating byte counts.
* Depending on the specific use case, if withEnoughBytesToFlush also meets the performance requirements, then each flush will have a more
* uniform data volume. The priority of withEnoughLinesToFlush is higher than withEnoughBytesToFlush.
*/
client.withEnoughLinesToFlush(10000);
/*
* MxClient has added the following API to adjust the number of Tuples for each CSV conversion sub - task.
* For example, if a flush has a total of 10,000 Tuples, and the following API sets BatchSize to 2000,
* then the CSV conversion of these 10,000 Tuples will be split into 5 sub - tasks, each handling 2000 Tuples and executed concurrently.
* (Supported from v1.1.0)
*/
client.withCSVConstructionBatchSize(2000);
/*
* Each MxClient has its own private object pool.
* When using it, you need to set the object pool size reasonably based on the number of Tuples flushed by each MxClient each time.
*/
// client.useTuplesPool(poolSize);
/*
* MxClient supports compression and needs to be used with mxgate v4.7.6 and higher versions.
*/
// client.withCompress();
/*
* For HTTP requests, base64 encoding is not required. For gRPC, base64 encoding is necessary.
*/
// client.withBase64Encode4Compress();
/*
* MxClient can register a DataPostListener. The success and failure of each batch of data sent will be
* called back in onSuccess and onFailure, respectively. You can understand which Tuples were successfully written and which failed.
*/
client.registerDataPostListener(new DataPostListener() {
@Override
public void onSuccess(Result result) {
System.out.println(CUSTOMER_LOG_TAG + "Send tuples success: " + result.getMsg());
System.out.println(CUSTOMER_LOG_TAG + "Succeed lines onSuccess callback " + result.getSucceedLines());
}
@Override
public void onFailure(Result result) {
/*
* result.getErrorTuplesMap() contains a map of error lines and their reasons Tuple -> String.
* Key is the error line, and value is the reason.
*/
for (Map.Entry<Tuple, String> entry : result.getErrorTuplesMap().entrySet()) {
l.error(CUSTOMER_LOG_TAG + "error tuple of table={}, tuple={}, reason={}", entry.getKey().getTableName(), entry.getKey(), entry.getValue());
for (Column c : entry.getKey().getColumns()) {
l.error(CUSTOMER_LOG_TAG + "error entry columns {} = {}", c.getColumnName(), c.getValue());
}
}
System.out.println(result.getSuccessLines());
}
});
}
}Obtain an empty Tuple object for assignment using the generateEmptyTuple API of MxClient.
Tuple tuple1 = client.generateEmptyTuple();
Tuple tuple2 = client.generateEmptyTuple();Assign values to the empty Tuple in a Key - Value manner. The Key is the column name of the corresponding database table; the Value is the value of that field. If some fields are allowed to be null or have default values, you can omit the values for those fields, and the SDK will automatically fill in the default or null values.
The Tuple obtained through this API no longer maintains the database table metadata internally and is a more lightweight Tuple. Therefore, when you call MxClient.appendTuple() to append this lightweight Tuple to MxClient, it skips the data validity check based on the database table metadata, allowing the Tuple data to be flushed at the fastest speed.
// Tuple tuple = client.generateEmptyTupleLite(); // (Supported from v1.1.0, uncomment this line to use)Note!
When adding columns to this lightweight Tuple, you need to manually maintain the order ofaddColumn() key -> valuepairs to ensure that it corresponds exactly to the order of columns in the database table.
For example, if the table column order is as follows:
Column | Type | Collation | Nullable | Default
--------+-----------------------------+-----------+----------+---------
ts | timestamp without time zone | | |
tag | integer | | not null |
c1 | double precision | | |
c2 | double precision | | |
c3 | double precision | | |
c4 | double precision | | |
c5 | text | | |
c6 | text | | |
c7 | text | | |
c8 | text | | |Then the order of adding columns must match the order of columns in the database table, as shown in the example below:
tuple1.addColumn("ts", "2022-05-18 16:30:06");
tuple1.addColumn("tag", 102020030);
tuple1.addColumn("c1", 1.1);
tuple1.addColumn("c2", 2.2);
tuple1.addColumn("c3", 3.3);
tuple1.addColumn("c4", 4.4);
tuple1.addColumn("c5", "Chinese character test - 1");
tuple1.addColumn("c6", "lTxFCVLwcDTKbNbjau_c6");
tuple1.addColumn("c7", "lTxFCVLwcDTKbNbjau_c7");
tuple1.addColumn("c8", "lTxFCVLwcDTKbNbjau_c8");
tuple2.addColumn("ts", "2022-05-18 16:30:06");
tuple2.addColumn("tag", 102020030);
tuple2.addColumn("c1", 1.1);
tuple2.addColumn("c2", 2.2);
tuple2.addColumn("c3", 3.3);
tuple2.addColumn("c4", 4.4);
tuple2.addColumn("c5", "Chinese character test - 2");
tuple2.addColumn("c6", "lTxFCVLwcDTKbNbjau_c26");
tuple2.addColumn("c7", "lTxFCVLwcDTKbNbjau_c27");
tuple2.addColumn("c8", "lTxFCVLwcDTKbNbjau_c28");Finally, append the assigned Tuple to MxClient.
client.appendTuples(tuple1, tuple2);Note!
MxClient provides multiple APIs that allow you to append one Tuple at a time, multiple Tuples at once, or a List of Tuples. For example:client.appendTuple();,client.appendTupleList();. You can also call theflush()method of MxClient to manually send the data, regardless of how many Tuples have been written, such asclient.flush();.
Synchronous sending means sending Tuples to mxgate synchronously for subsequent processing (such as committing Kafka offsets).
Tuple tuple1 = client.generateEmptyTuple();
tuple1.addColumn("ts", "2022-05-18 16:30:06");
tuple1.addColumn("tag", 102020030);
tuple1.addColumn("c1", 1.1);
tuple1.addColumn("c2", 2.2);
tuple1.addColumn("c3", 3.3);
tuple1.addColumn("c4", 4.4);
tuple1.addColumn("c5", "lTxFCVLwcDTKbNbjau_c5");
tuple1.addColumn("c6", "lTxFCVLwcDTKbNbjau_c6");
tuple1.addColumn("c7", "lTxFCVLwcDTKbNbjau_c7");
tuple1.addColumn("c8", "lTxFCVLwcDTKbNbjau_c8");
/*
* appendTupleBlocking returns a boolean value:
* true: The MxClient has written the set bytes size and can send a request;
* false: The set bytes size has not been reached yet.
*/
try {
if (client.appendTupleBlocking(tuple1)) {
l.info("append tuples enough");
// Manually trigger tuples flush
client.flushBlocking();
}
/*
* If the following flushBlocking throws an exception, it means that the entire batch of Tuples failed to write to mxgate. The caller can handle this exception accordingly.
*/
} catch (AllTuplesFailException e) {
l.error("Tuples fail and catch the exception return.", e);
for (Map.Entry<Tuple, String> entry : result.getErrorTuplesMap().entrySet()) {
l.error(CUSTOMER_LOG_TAG + "error tuple of table={}, tuple={}, reason={}", entry.getKey().getTableName(), entry.getKey(), entry.getValue());
for (Column c : entry.getKey().getColumns()) {
l.error(CUSTOMER_LOG_TAG + "error entry columns {} = {}", c.getColumnName(), c.getValue());
}
}
}
/*
* If the following flushBlocking throws an exception, it means that some Tuples failed to write to mxgate. The caller can handle this exception accordingly.
*/
} catch (PartiallyTuplesFailException e) {
for (Map.Entry<Tuple, String> entry : result.getErrorTuplesMap().entrySet()) {
l.error(CUSTOMER_LOG_TAG + "error tuple of table={}, tuple={}, reason={}", entry.getKey().getTableName(), entry.getKey(), entry.getValue());
for (Column c : entry.getKey().getColumns()) {
l.error(CUSTOMER_LOG_TAG + "error entry columns {} = {}", c.getColumnName(), c.getValue());
}
}
}import java.io.BufferedReader;
import java.io.DataOutputStream;
import java.io.InputStreamReader;
import java.net.HttpURLConnection;
import java.net.URL;
public class MxgateExample {
public static void main(String[] args) throws Exception {
MxgateExample http = new MxgateExample();
http.sendingPostRequest();
}
/*
* HTTP Post request
*/
private void sendingPostRequest() throws Exception {
/*
* mxgate listens on port 8086 of localhost
*/
String url = "http://localhost:8086/";
URL obj = new URL(url);
HttpURLConnection con = (HttpURLConnection) obj.openConnection();
/*
* Setting basic post request
*/
con.setRequestMethod("POST");
con.setRequestProperty("Content-Type","text/plain");
String postJsonData = "public.testtable\n1603777821|1|101|201|301\n1603777822|2|102|202|302\n1603777823|3|103|203|303";
con.setDoOutput(true);
DataOutputStream wr = new DataOutputStream(con.getOutputStream());
/*
* When data contains Chinese characters, you can encode it using postJsonData.getBytes("UTF-8")
*/
wr.write(postJsonData.toString().getBytes("UTF-8"));
wr.flush();
wr.close();
int responseCode = con.getResponseCode();
System.out.println("Sending 'POST' request to URL : " + url);
System.out.println("Post Data : " + postJsonData);
System.out.println("Response Code : " + responseCode);
BufferedReader in = new BufferedReader(new InputStreamReader(con.getInputStream()));
String output;
StringBuffer response = new StringBuffer();
while ((output = in.readLine()) != null) {
response.append(output);
}
in.close();
System.out.println(response.toString());
}
}import http.client
class MxgateExample(object):
def __init__(self):
/*
* mxgate listens on port 8086 of localhost
*/
self.url = "localhost:8086"
self.postData = "public.testtable\n/" \
"1603777821|1|101|201|301\n/" \
"1603777822|2|102|202|302\n/" \
"1603777823|3|103|203|303"
self.headers = {"Content-Type": "text/plain"}
/*
* HTTP Post request
*/
def sending_post_request(self):
conn = http.client.HTTPConnection(self.url)
conn.request("POST", "/", self.postData, self.headers)
response = conn.getresponse()
response_code = response.getcode()
print(f"Sending 'POST' request to URL : {self.url}")
print(f"Post Data : {self.postData}")
print(f"Response Code : {response_code}")
output = response.read()
print(output)
if __name__ == '__main__':
gate_post = MxgateExample()
gate_post.sending_post_request()It is recommended to use the C# Core development environment for development.
using System;
using System.IO;
using System.Net;
using System.Text;
namespace HttpPostTest
{
class Program
{
static void Main(string[] args)
{
var url = "http://10.13.2.177:8086/";
var txt = "public.dest\n2021-01-01 00:00:00,1,a1\n2021-01-01 00:00:00,2,a2\n2021-01-01 00:00:00,3,a3";
HttpPost(url,txt);
}
public static string HttpPost(string url, string content){
string result = "";
HttpWebRequest req = (HttpWebRequest)WebRequest.Create(url);
req.Method = "POST";
req.ContentType = "text/plain";
/*
* region Add Post parameters
*/
byte[] data = Encoding.UTF8.GetBytes(content);
req.ContentLength = data.Length;
using (Stream reqStream = req.GetRequestStream()){
reqStream.Write(data, 0, data.Length);
reqStream.Close();
}
HttpWebResponse resp = (HttpWebResponse)req.GetResponse();
Stream stream = resp.GetResponseStream();
/*
* Get the response content
*/
using (StreamReader reader = new StreamReader(stream, Encoding.UTF8)){
result = reader.ReadToEnd();
}
return result;
}
}
}If you encounter the error "error when serving connection ***** body size exceeds the given limit", you need to increase the
max-body-bytesinmxgate.conf.
package main
import (
"bytes"
"net/http"
)
func PostDataToServer(URL string) error {
data := `public.testtable
1603777821|1|101|201|301
1603777822|2|102|202|302
1603777823|3|103|203|303
`
resp, err := http.Post(URL, "application/text", bytes.NewBuffer([]byte(data)))
if err != nil {
return err
}
if resp.StatusCode != 200 {
/*
* Deal with the response body.
*/
return nil
}
/*
* Deal with the response body.
*/
return nil
}
func main() {
err := PostDataToServer("http://127.0.0.1:8086")
if err != nil{
panic(err)
}
}Note!
For more MatrixGate features, refer to Tool Guide - mxgate.