reader

Documentation

Overview

Package reader contains logic for reading from a provider. Any types that implements the DataReader interface can be used in this system. The Result should provide a byte slice that is JSON unmarshallable, otherwise the data will be rejected.

Important Notes

When the token's context is cancelled, the reader should finish its job and return. The Time should be set when the data is read from the endpoint, otherwise it will lose its meaning. The engine will issue jobs based on the Interval, which is set in the configuration file.

Index

• Variables
• func WithBackoff(backoff int) func(Constructor) error
• func WithEndpoint(endpoint string) func(Constructor) error
• func WithInterval(interval time.Duration) func(Constructor) error
• func WithLogger(log tools.FieldLogger) func(Constructor) error
• func WithMapper(mapper datatype.Mapper) func(Constructor) error
• func WithName(name string) func(Constructor) error
• func WithTimeout(timeout time.Duration) func(Constructor) error
• func WithTypeName(typeName string) func(Constructor) error
• type Constructor
• type DataReader
• type EndpointNotAvailableError
  • func (e EndpointNotAvailableError) Error() string
• type InvalidEndpointError
  • func (e InvalidEndpointError) Error() string
• type LowBackoffValueError
  • func (e LowBackoffValueError) Error() string
• type LowIntervalError
  • func (e LowIntervalError) Error() string
• type LowTimeoutError
  • func (e LowTimeoutError) Error() string
• type Result

Examples

• DataReader (Read)
• WithEndpoint
• WithLogger
• WithName

Variables

var (
	ErrEmptyName       = fmt.Errorf("name cannot be empty")
	ErrEmptyEndpoint   = fmt.Errorf("endpoint cannot be empty")
	ErrEmptyTypeName   = fmt.Errorf("type_name cannot be empty")
	ErrBackoffExceeded = fmt.Errorf("endpoint gone too long")
	ErrPingNotCalled   = fmt.Errorf("the caller forgot to ask me pinging")
	ErrInvalidJSON     = fmt.Errorf("payload is invalid JSON object")
	ErrNillLogger      = fmt.Errorf("nil logger")
)

Errors regarding reading from an endpoint. ErrBackoffExceeded is the error when the endpoint's absence has exceeded the backoff value. It is not strictly an error, it is however a pointer to an error in the past.

Functions

func WithBackoff

func WithBackoff(backoff int) func(Constructor) error

WithBackoff sets the backoff of the reader.

func WithEndpoint

func WithEndpoint(endpoint string) func(Constructor) error

WithEndpoint sets the endpoint of the reader.

Example

r := reader.Reader{}
err := WithEndpoint("")(&r)
err = errors.Cause(err)
fmt.Println("Error:", err == ErrEmptyEndpoint)

err = WithEndpoint("somewhere")(&r)
err = errors.Cause(err)
fmt.Println("Error:", err)

err = WithEndpoint("http://localhost")(&r)
fmt.Println("Error:", err == nil)

Output:

Error: true
Error: invalid endpoint: somewhere
Error: true

func WithInterval

func WithInterval(interval time.Duration) func(Constructor) error

WithInterval sets the interval of the reader.

func WithLogger

func WithLogger(log tools.FieldLogger) func(Constructor) error

WithLogger sets the log of the reader.

Example

r := reader.Reader{}
err := WithLogger(nil)(&r)
fmt.Println("Error:", err == ErrNillLogger)

err = WithLogger(tools.DiscardLogger())(&r)
fmt.Println("Error:", err == nil)

Output:

Error: true
Error: true

func WithMapper

func WithMapper(mapper datatype.Mapper) func(Constructor) error

WithMapper sets the mapper of the reader.

func WithName

func WithName(name string) func(Constructor) error

WithName sets the name of the reader.

Example

r := reader.Reader{}
err := WithName("")(&r)
fmt.Println("Error:", err == ErrEmptyName)

err = WithName("some name")(&r)
fmt.Println("Error:", err == nil)

Output:

Error: true
Error: true

func WithTimeout

func WithTimeout(timeout time.Duration) func(Constructor) error

WithTimeout sets the timeout of the reader.

func WithTypeName

func WithTypeName(typeName string) func(Constructor) error

WithTypeName sets the typeName of the reader.

Types

type Constructor

type Constructor interface {
	SetLogger(logger tools.FieldLogger)
	SetName(name string)
	SetTypeName(typeName string)
	SetEndpoint(endpoint string)
	SetMapper(mapper datatype.Mapper)
	SetInterval(interval time.Duration)
	SetTimeout(timeout time.Duration)
	SetBackoff(backoff int)
}

Constructor is a Reader object that will accept configurations.

type DataReader

type DataReader interface {
	Name() string
	Read(*token.Context) (*Result, error)
	Ping() error
	Mapper() datatype.Mapper
	TypeName() string
	Timeout() time.Duration
	Interval() time.Duration
	Endpoint() string
	Backoff() int
}

DataReader receives job requests to read from the target. It returns an error if the data cannot be read or the connection is refused.

Notes

Readers should not intercept the engine's decision on the TypeName, unless they have a valid reason. Ping should ping the endpoint and return nil if was successful. The Engine will not launch the reader if the ping result is an error. When the context is timed-out or cancelled, Read should return. The Engine uses returned object from Mapper to present the data to recorders. TypeName is usually the application name and is set by the user in the configuration file.

Example (Read)

This example shows the reader hits the endpoint when the Read method is called.

package main

import (
	"context"
	"fmt"
	"io"
	"net/http"
	"net/http/httptest"

	reader "github.com/arsham/expipe/reader/testing"
	"github.com/arsham/expipe/tools/token"
)

func main() {

	ts := httptest.NewServer(http.HandlerFunc(
		func(w http.ResponseWriter, r *http.Request) {
			io.WriteString(w, `{"the key": "is the value!"}`)
		}),
	)
	defer ts.Close()

	// This reader is a mocked version,
	// but the example's principals stays the same.
	red := reader.GetReader(ts.URL)
	err := red.Ping()
	fmt.Println("Ping errors:", err)

	job := token.New(context.Background())
	res, err := red.Read(job) // Issuing a job

	if err == nil { // Lets check the errors
		fmt.Println("No errors reported")
	}

	fmt.Println("Result is:", string(res.Content))

}

Output:

Ping errors: <nil>
No errors reported
Result is: {"the key": "is the value!"}

type EndpointNotAvailableError

type EndpointNotAvailableError struct {
	Endpoint string
	Err      error
}

EndpointNotAvailableError is the error when the endpoint is not available.

func (EndpointNotAvailableError) Error

func (e EndpointNotAvailableError) Error() string

type InvalidEndpointError

type InvalidEndpointError string

InvalidEndpointError is the error when the endpoint is not a valid url.

func (InvalidEndpointError) Error

func (e InvalidEndpointError) Error() string

type LowBackoffValueError

type LowBackoffValueError int64

LowBackoffValueError is the error when the backoff value is lower than 5

func (LowBackoffValueError) Error

func (e LowBackoffValueError) Error() string

type LowIntervalError

type LowIntervalError time.Duration

LowIntervalError is the error when the interval is zero

func (LowIntervalError) Error

func (e LowIntervalError) Error() string

type LowTimeoutError

type LowTimeoutError time.Duration

LowTimeoutError is the error when the interval is zero

func (LowTimeoutError) Error

func (e LowTimeoutError) Error() string

type Result

type Result struct {
	// ID is the job ID given by the Engine.
	// This ID should not be changed until it is recorded.
	ID token.ID

	//Time is set after the request was successfully read.
	Time time.Time

	// TypeName comes from the configuration, but the Engine might decide
	// to change it.
	TypeName string

	// Content should be json unmarshallable, otherwise the job will be dropped.
	Content []byte

	// Mapper is the mapper set in the reader.
	Mapper datatype.Mapper
}

Result is constructed every time a new data is fetched.