ais

README

Conduit Connector for SPIRE-AIS GraphQL API

Conduit for Spire Martime 2.0 API.

How to build?

Run make build to build the connector.

Testing

Run make test to run all the unit tests. Run make test-integration to run the integration tests.

The Docker compose file at test/docker-compose.yml can be used to run the required resource locally.

Source

The source connector pulls data from Spire's Maritime 2.0 GraphQL API

Configuration

A Spire API token is required to use this connector

name	description	required	default value
apiUrl	Spire API URL to use for accessing the Maritime 2.0 GraphQL API.	false	https://api.spire.com/graphql
token	Access token to use when accessing the Spire GraphQL API.	true
query	The query to send to the Spire GraphQL API.	false	Default graphQL Query is in query.go
batchSize	The maximum number of results to retrieve from the Spire GraphQL API for each request.	false	100

Known Issues & Limitations

• There's currently no pre-flight validation on the GraphQL query

Documentation

Index

• Constants
• Variables
• func NewSource() sdk.Source
• func Specification() sdk.Specification
• type Config
• type CurrentVoyage
• type Dimensions
• type GraphQLClient
• type Iterator
  • func NewIterator(client GraphQLClient, token string, query string, batchSize int, ...) (*Iterator, error)
  • func (it *Iterator) HasNext(ctx context.Context) bool
  • func (it *Iterator) Next(ctx context.Context) (opencdc.Record, error)
• type IteratorCreator
• type IteratorInterface
• type LastPositionUpdate
• type Node
• type PageInfo
• type Source
  • func (s *Source) Ack(ctx context.Context, position opencdc.Position) error
  • func (s *Source) Configure(ctx context.Context, cfg config.Config) error
  • func (s *Source) GetConfig() SourceConfig
  • func (s *Source) Open(ctx context.Context, pos opencdc.Position) error
  • func (s *Source) Parameters() config.Parameters
  • func (s *Source) Read(ctx context.Context) (opencdc.Record, error)
  • func (s *Source) Teardown(ctx context.Context) error
• type SourceConfig
  • func (SourceConfig) Parameters() map[string]config.Parameter
• type SourceIteratorCreator
  • func (ic SourceIteratorCreator) NewIterator(client GraphQLClient, token string, query string, batchSize int, ...) (*Iterator, error)
• type StaticData
• type TotalCount
• type Vessels

Constants

const (
	SourceConfigApiUrl    = "apiUrl"
	SourceConfigBatchSize = "batchSize"
	SourceConfigQuery     = "query"
	SourceConfigToken     = "token"
)

Variables

var Connector = sdk.Connector{
	NewSpecification: Specification,
	NewSource:        NewSource,
	NewDestination:   nil,
}

Connector combines all constructors for each plugin in one struct.

Functions

func NewSource

func NewSource() sdk.Source

func Specification

func Specification() sdk.Specification

Specification returns the connector's specification.

Types

type Config

type Config struct {
	// APIURL is the Spire API URL to use for accessing the Spire GraphQL API.
	APIURL string `json:"apiUrl" default:"https://api.spire.com/graphql"`

	// Token is the access token to use when accessing the Spire GraphQL API.
	Token string `json:"token" validate:"required"`

	// batchSize is the quantity of vessels to retrieve per Spire GraphQL API call.
	BatchSize int `json:"batchSize" default:"100"`
}

Config contains shared config parameters, common to the source and destination. If you don't need shared parameters you can entirely remove this file.

type CurrentVoyage

type CurrentVoyage struct {
	Destination     string  `json:"destination"`
	Draught         float64 `json:"draught"`
	ETA             string  `json:"eta"`
	Timestamp       string  `json:"timestamp"`
	UpdateTimestamp string  `json:"updateTimestamp"`
}

type Dimensions

type Dimensions struct {
	A      float64 `json:"a"`
	B      float64 `json:"b"`
	C      float64 `json:"c"`
	D      float64 `json:"d"`
	Width  float64 `json:"width"`
	Length float64 `json:"length"`
}

type GraphQLClient

type GraphQLClient interface {
	Run(ctx context.Context, req *graphql.Request, resp interface{}) error
}

Add GraphQLClient interface for dependency injection

type Iterator

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

Updated Iterator struct with logger and client dependencies

func NewIterator

func NewIterator(client GraphQLClient, token string, query string, batchSize int, p opencdc.Position) (*Iterator, error)

func (*Iterator) HasNext

func (it *Iterator) HasNext(ctx context.Context) bool

func (*Iterator) Next

func (it *Iterator) Next(ctx context.Context) (opencdc.Record, error)

type IteratorCreator

type IteratorCreator interface {
	NewIterator(client GraphQLClient, token string, query string, batchSize int, p opencdc.Position) (*Iterator, error)
}

type IteratorInterface

type IteratorInterface interface {
	HasNext(ctx context.Context) bool
	Next(ctx context.Context) (opencdc.Record, error)
}

type LastPositionUpdate

type LastPositionUpdate struct {
	Accuracy           string  `json:"accuracy"`
	CollectionType     string  `json:"collectionType"`
	Course             float64 `json:"course"`
	Heading            float64 `json:"heading"`
	Latitude           float64 `json:"latitude"`
	Longitude          float64 `json:"longitude"`
	Maneuver           string  `json:"maneuver"`
	NavigationalStatus string  `json:"navigationalStatus"`
	Rot                float64 `json:"rot"`
	Speed              float64 `json:"speed"`
	Timestamp          string  `json:"timestamp"`
	UpdateTimestamp    string  `json:"updateTimestamp"`
}

type Node

type Node struct {
	ID                 string             `json:"id"`
	UpdateTimestamp    string             `json:"updateTimestamp"`
	StaticData         StaticData         `json:"staticData"`
	LastPositionUpdate LastPositionUpdate `json:"lastPositionUpdate"`
	CurrentVoyage      CurrentVoyage      `json:"currentVoyage"`
}

type PageInfo

type PageInfo struct {
	HasNextPage bool   `json:"hasNextPage"`
	EndCursor   string `json:"endCursor"`
}

type Source

type Source struct {
	sdk.UnimplementedSource
	// contains filtered or unexported fields
}

func (*Source) Ack

func (s *Source) Ack(ctx context.Context, position opencdc.Position) error

func (*Source) Configure

func (s *Source) Configure(ctx context.Context, cfg config.Config) error

func (*Source) GetConfig

func (s *Source) GetConfig() SourceConfig

func (*Source) Open

func (s *Source) Open(ctx context.Context, pos opencdc.Position) error

func (*Source) Parameters

func (s *Source) Parameters() config.Parameters

func (*Source) Read

func (s *Source) Read(ctx context.Context) (opencdc.Record, error)

func (*Source) Teardown

func (s *Source) Teardown(ctx context.Context) error

type SourceConfig

type SourceConfig struct {
	// Config includes parameters that are the same in the source and destination.
	Config

	// Query is the GraphQL Query to use when pulling data from the Spire API.
	Query string `json:"query"`
}

func (SourceConfig) Parameters

func (SourceConfig) Parameters() map[string]config.Parameter

type SourceIteratorCreator

type SourceIteratorCreator struct {
}

func (SourceIteratorCreator) NewIterator

func (ic SourceIteratorCreator) NewIterator(client GraphQLClient, token string, query string, batchSize int, p opencdc.Position) (*Iterator, error)

type StaticData

type StaticData struct {
	AisClass        string     `json:"aisClass"`
	Flag            string     `json:"flag"`
	Name            string     `json:"name"`
	Callsign        string     `json:"callsign"`
	Timestamp       string     `json:"timestamp"`
	UpdateTimestamp string     `json:"updateTimestamp"`
	ShipType        string     `json:"shipType"`
	ShipSubType     string     `json:"shipSubType"`
	MMSI            int        `json:"mmsi"`
	IMO             int        `json:"imo"`
	Dimensions      Dimensions `json:"dimensions"`
}

type TotalCount

type TotalCount struct {
	Value    int    `json:"value"`
	Relation string `json:"string"`
}

type Vessels

type Vessels struct {
	PageInfo   PageInfo   `json:"pageInfo"`
	TotalCount TotalCount `json:"totalCount"`
	Nodes      []Node     `json:"nodes"`
}