ingest

Documentation

Overview

Package ingest implements the IndyKite Ingest Service API Client.

Index

• type Client
  • func NewClient(ctx context.Context, opts ...api.ClientOption) (*Client, error)
  • func NewTestClient(client ingestpb.IngestAPIClient) (*Client, error)
  • func (c *Client) Close() error
  • func (c *Client) CloseStream() error
  • func (c *Client) IngestRecord(ctx context.Context, record *ingestpb.Record, opts ...grpc.CallOption) (*ingestpb.IngestRecordResponse, error)
  • func (c *Client) OpenStreamClient(ctx context.Context) error
  • func (c *Client) ReceiveResponse() (*ingestpb.StreamRecordsResponse, error)
  • func (c *Client) SendRecord(record *ingestpb.Record) error
  • func (c *Client) StreamRecords(records []*ingestpb.Record) ([]*ingestpb.StreamRecordsResponse, error)

Examples

• Client.IngestRecord
• Client.StreamRecords
• NewClient (Default)
• NewClient (Options)

Types

type Client

type Client struct {
	// contains filtered or unexported fields
}

func NewClient

func NewClient(ctx context.Context, opts ...api.ClientOption) (*Client, error)

NewClient creates a new Ingest API gRPC Client.

Example (Default)

This example demonstrates how to create a new Ingest Client.

package main

import (
	"context"
	"log"

	"github.com/indykite/indykite-sdk-go/ingest"
)

func main() {
	client, err := ingest.NewClient(context.Background())
	if err != nil {
		log.Fatalf("failed to create client %v", err)
	}
	defer func() {
		_ = client.Close()
	}()
}

Example (Options)

This example demonstrates how to create a new Ingest Client.

package main

import (
	"context"
	"log"

	api "github.com/indykite/indykite-sdk-go/grpc"
	"github.com/indykite/indykite-sdk-go/ingest"
)

func main() {
	client, err := ingest.NewClient(context.Background(),
		api.WithCredentialsJSON([]byte(`{}`)))
	if err != nil {
		log.Fatalf("failed to create client %v", err)
	}
	defer func() {
		_ = client.Close()
	}()
}

func NewTestClient

func NewTestClient(client ingestpb.IngestAPIClient) (*Client, error)

NewTestClient creates a new Ingest gRPC Client.

func (*Client) Close

func (c *Client) Close() error

Close closes the connection to the API service. The user should invoke this when the client is no longer required.

func (*Client) CloseStream

func (c *Client) CloseStream() error

CloseStream closes the gRPC stream.

func (*Client) IngestRecord

func (c *Client) IngestRecord(
	ctx context.Context,
	record *ingestpb.Record,
	opts ...grpc.CallOption,
) (*ingestpb.IngestRecordResponse, error)

Example

This example demonstrates how to use the Ingest Client to ingest a single record.

client, err := ingest.NewClient(context.Background())
if err != nil {
	log.Fatalf("failed to create client %v", err)
}
defer func() {
	_ = client.Close()
}()

response, err := client.IngestRecord(context.Background(), record1)
if err != nil {
	log.Fatalf("failed to invoke operation on IndyKite Client %v", err) //nolint:gocritic
}
json := protojson.MarshalOptions{
	Multiline: true,
}
fmt.Println(json.Format(response))

func (*Client) OpenStreamClient

func (c *Client) OpenStreamClient(ctx context.Context) error

OpenStreamClient opens a stream, ready to ingest records.

func (*Client) ReceiveResponse

func (c *Client) ReceiveResponse() (*ingestpb.StreamRecordsResponse, error)

ReceiveResponse returns the next response available on the stream.

func (*Client) SendRecord

func (c *Client) SendRecord(record *ingestpb.Record) error

SendRecord sends a record on the opened stream.

func (*Client) StreamRecords

func (c *Client) StreamRecords(records []*ingestpb.Record) (
	[]*ingestpb.StreamRecordsResponse,
	error)

StreamRecords is a helper that takes a slice of records and handles opening the stream, sending the records, getting the responses, and closing the stream.

Example

This example demonstrates how to use the Ingest Client to stream multiple records.

client, err := ingest.NewClient(context.Background())
if err != nil {
	log.Fatalf("failed to create client %v", err)
}
defer func() {
	_ = client.Close()
}()

ctx, cancel := context.WithTimeout(context.Background(), 120*time.Second)
defer cancel()

err = client.OpenStreamClient(ctx)
if err != nil {
	log.Fatalf("failed to open stream %v", err) //nolint:gocritic
}

for _, record := range []*ingestv2pb.Record{record1, record2} {
	err = client.SendRecord(record)
	if err != nil {
		log.Fatalf("failed to send record on stream %v", err)
	}
}

for {
	resp, err := client.ReceiveResponse()
	if err != nil {
		if err.Error() == "EOF" {
			break
		}
		log.Fatalf("failed to receive responses %v", err)
	}
	json := protojson.MarshalOptions{
		Multiline: true,
	}
	fmt.Println(json.Format(resp))
}