Golang SDK

Create real-time synchronization task

Create a task on the Dashboard or with the command line tool, and set Auto Push (autonomous push) as the data source type.

Import Golang SDK

To use the Golang SDK, you need to import it into your projects. Import the Golang SDK:

import "github.com/apache/inlong/inlong-sdk/dataproxy-sdk-twins/dataproxy-sdk-golang/dataproxy"

Data report process

After importing the SDK, you can initialize a Client instance, and then call Send() or SendAsync() to produce messages to DataProxy, SDK will put the messages with a same StreamID into a batch and send them together. A demo can be found at: example_test.go. Basically, there are 3 steps to produce messages:

Initialize the SDK

client, err := dataproxy.NewClient(
    dataproxy.WithGroupID("test"),              
dataproxy.WithURL("http://127.0.0.1:8083/inlong/manager/openapi/dataproxy/getIpList"),
    dataproxy.WithMetricsName("test"),
)
if err != nil {
    fmt.Println(err)
    return
}

Parameters:

• dataproxy.WithGroupID("test") sets the GroupID;
• dataproxy.WithURL("http://127.0.0.1:8083/inlong/manager/openapi/dataproxy/getIpList") sets the Manager URL;
• dataproxy.WithMetricsName("test") sets the value of the metrics label: “name” of the Client.

Call the Send/SendAsync method to send

The send methods of the SDK are goroutine safe, you can send a message synchronously or asynchronously, there are both synchronous and asynchronous examples in the demo.

Send synchronously :

for i := 0; i < 1000; i++ {
    err := client.Send(context.Background(), dataproxy.Message{
        GroupID:  "test",
        StreamID: "test",
        Payload:  []byte("test|a|b|c"),
    })
    if err !=nil {
        fmt.Println(err)
    }
}

Send asynchronously:

var success atomic.Uint64
var failed atomic.Uint64
for i := 0; i < 1000; i++ {
    client.SendAsync(context.Background(),
        dataproxy.Message{
            GroupID:  "test",
            StreamID: "test",
            Payload:  []byte("test|a|b|c"),
        },
        func(msg dataproxy.Message, err error) {
            if err != nil {
                success.Add(1)
            } else {
                failed.Add(1)
            }
        })
}
// wait async send finish
time.Sleep(3 * time.Second)
fmt.Println("success:", success.Load())
fmt.Println("failed:", failed.Load())

The asynchronous way is recommended, as the synchronous way has no concurrency, messages can be sent after the previous one is done or timeout, it can’t fulfill your requirements when you need high throughput.

Close the SDK

Closing the SDK can be done simply by calling the Close() method of the Client:

client.Close()

Notes

• GroupID and URL are mandatory options when you initialize the SDK, you can call dataproxy.WithGroupID() and dataproxy.WithURL() to set them;
• When you initialize more the one instance of Client, the MetricsName must be set to different values, or it will be failed when pulling metrics, you can call dataproxy.WithMetricsName() to set it;
• The default values of the config options of the SDK are good enough for production, you don’t need the change then unless you really need to do that, the default values are described in a section below;
• The Send() method of the SDK has no concurrent, messages are sent one by one, it is NOT recommended;
• The SDK will retry 2 times each message, if it still failed finally, it is up to you to decide what to do next.

Errors

Some common errors:

Config options

options refers to options.go

Metrics

metrics refers to metrics.go

error codes refers to worker.go