ksqldb

Documentation

Index

• Constants
• type BasicAuth
• type Client
  • func NewClient(url string, options ...func(*Client)) *Client
  • func (c *Client) CloseQuery(ctx context.Context, queryId string) error
  • func (c *Client) Pull(ctx context.Context, q *QueryRequest) (*QueryResponse, []Row, error)
  • func (c *Client) Push(ctx context.Context, q *QueryRequest, cb func(qr *QueryResponse, row Row)) error
• type QueryRequest
• type QueryResponse
• type Row
  • func (r Row) Bool(idx int) bool
  • func (r Row) Float64(idx int) float64
  • func (r Row) Int(idx int) int
  • func (r Row) String(idx int) string
  • func (r Row) Time(idx int) *time.Time

Constants

const (

	// Default response content type for pull & push queries
	// In the case of a successful query, if the content type is application/vnd.ksqlapi.delimited.v1,
	// the results are returned as a header JSON object followed by zero or more JSON arrays that are delimited by newlines.
	ContentTypeDelim = "application/vnd.ksqlapi.delimited.v1; charset=utf-8"

	// Default serialization format for requests and responses.
	ContentTypeDefault = "application/vnd.ksql.v1+json; charset=utf-8"

	// EndpointRunStreamQuery is used to run push and pull queries.
	// These endpoints are only available when using HTTP 2.
	EndpointRunStreamQuery string = "/query-stream"
	// EndpointCloseQuery used to terminates a push query.
	EndpointCloseQuery string = "/close-query"
)

Types

type BasicAuth

type BasicAuth struct {
	Username string
	Password string
}

type Client

type Client struct {
	HTTPClient *http.Client
	BasicAuth  *BasicAuth
	// contains filtered or unexported fields
}

func NewClient

func NewClient(url string, options ...func(*Client)) *Client

NewClient creates a new ksqlDB client.

func (*Client) CloseQuery

func (c *Client) CloseQuery(ctx context.Context, queryId string) error

CloseQuery terminates a query.

func (*Client) Pull

func (c *Client) Pull(ctx context.Context, q *QueryRequest) (*QueryResponse, []Row, error)

func (*Client) Push

func (c *Client) Push(ctx context.Context, q *QueryRequest, cb func(qr *QueryResponse, row Row)) error

type QueryRequest

type QueryRequest struct {
	SQL        string            `json:"sql"`
	Properties map[string]string `json:"streamsProperties,omitempty"`
}

type QueryResponse

type QueryResponse struct {
	QueryID     string   `json:"queryID"`
	ColumnNames []string `json:"columnNames"`
	ColumnTypes []string `json:"columnTypes"`
}

type Row

type Row []interface{}

func (Row) Bool

func (r Row) Bool(idx int) bool

func (Row) Float64

func (r Row) Float64(idx int) float64

func (Row) Int

func (r Row) Int(idx int) int

func (Row) String

func (r Row) String(idx int) string

func (Row) Time

func (r Row) Time(idx int) *time.Time