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 SetBackoff(backoff int) func(Constructor) error
• func SetEndpoint(endpoint string) func(Constructor) error
• func SetInterval(interval time.Duration) func(Constructor) error
• func SetLogger(log internal.FieldLogger) func(Constructor) error
• func SetMapper(mapper datatype.Mapper) func(Constructor) error
• func SetName(name string) func(Constructor) error
• func SetTimeout(timeout time.Duration) func(Constructor) error
• func SetTypeName(typeName string) func(Constructor) error
• type Constructor
• type DataReader
• type ErrEndpointNotAvailable
  • func (e ErrEndpointNotAvailable) Error() string
• type ErrInvalidEndpoint
  • func (e ErrInvalidEndpoint) Error() string
• type ErrLowBackoffValue
  • func (e ErrLowBackoffValue) Error() string
• type ErrLowInterval
  • func (e ErrLowInterval) Error() string
• type ErrLowTimeout
  • func (e ErrLowTimeout) Error() string
• type Result

Examples

• DataReader (Read)

Variables

var (
	// ErrEmptyName is the error when the package name is empty.
	ErrEmptyName = fmt.Errorf("name cannot be empty")

	// ErrEmptyEndpoint is the error when the given endpoint is empty.
	ErrEmptyEndpoint = fmt.Errorf("endpoint cannot be empty")

	// ErrEmptyTypeName is the error when the type_name is an empty string.
	ErrEmptyTypeName = fmt.Errorf("type_name cannot be empty")

	// 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.
	ErrBackoffExceeded = fmt.Errorf("endpoint gone too long")

	// ErrPingNotCalled is the error if the caller calls the record without pinging.
	ErrPingNotCalled = fmt.Errorf("the caller forgot to ask me pinging")
)

Functions

func SetBackoff

func SetBackoff(backoff int) func(Constructor) error

SetBackoff sets the backoff of the reader

func SetEndpoint

func SetEndpoint(endpoint string) func(Constructor) error

SetEndpoint sets the endpoint of the reader

func SetInterval

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

SetInterval sets the interval of the reader

func SetLogger

func SetLogger(log internal.FieldLogger) func(Constructor) error

SetLogger sets the log of the reader

func SetMapper

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

SetMapper sets the mapper of the reader

func SetName

func SetName(name string) func(Constructor) error

SetName sets the name of the reader

func SetTimeout

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

SetTimeout sets the timeout of the reader

func SetTypeName

func SetTypeName(typeName string) func(Constructor) error

SetTypeName sets the typeName of the reader

Types

type Constructor

type Constructor interface {
	// SetLogger is for setting the Logger
	SetLogger(logger internal.FieldLogger)

	// SetName is for setting the Name
	SetName(name string)

	// SetTypeName is for setting the TypeName
	SetTypeName(typeName string)

	// SetEndpoint is for setting the Endpoint
	SetEndpoint(endpoint string)

	// SetMapper is for setting the SetMapper
	SetMapper(mapper datatype.Mapper)

	// SetInterval is for setting the Interval
	SetInterval(interval time.Duration)

	// SetTimeout is for setting the Timeout
	SetTimeout(timeout time.Duration)

	// SetBackoff is for setting the Backoff
	SetBackoff(backoff int)
}

Constructor is an interface for setting up an object for testing.

type DataReader

type DataReader interface {
	// Name should return the representation string for this reader.
	// Choose a very simple and unique name.
	Name() string

	// 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.
	Ping() error

	// When the context is timed-out or cancelled, the reader should return.
	Read(*token.Context) (*Result, error)

	// Mapper should return an instance of the datatype mapper.
	// Engine uses this object to present the data to recorders.
	Mapper() datatype.Mapper

	// TypeName is usually the application name and is set by the user in
	// the configuration file.
	TypeName() string

	// Timeout is required by the Engine so it can read the time-outs.
	Timeout() time.Duration

	// Interval is required by the Engine so it can read the intervals.
	Interval() time.Duration
}

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.

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"

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

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()

	red := reader.GetReader(ts.URL) // this is a mocked version, but the example's principals stays the same.
	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 ErrEndpointNotAvailable

type ErrEndpointNotAvailable struct {
	Endpoint string
	Err      error
}

ErrEndpointNotAvailable is the error when the endpoint is not available.

func (ErrEndpointNotAvailable) Error

func (e ErrEndpointNotAvailable) Error() string

type ErrInvalidEndpoint

type ErrInvalidEndpoint string

ErrInvalidEndpoint is the error when the endpoint is not a valid url

func (ErrInvalidEndpoint) Error

func (e ErrInvalidEndpoint) Error() string

type ErrLowBackoffValue

type ErrLowBackoffValue int64

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

func (ErrLowBackoffValue) Error

func (e ErrLowBackoffValue) Error() string

type ErrLowInterval

type ErrLowInterval time.Duration

ErrLowInterval is the error when the interval is zero

func (ErrLowInterval) Error

func (e ErrLowInterval) Error() string

type ErrLowTimeout

type ErrLowTimeout time.Duration

ErrLowTimeout is the error when the interval is zero

func (ErrLowTimeout) Error

func (e ErrLowTimeout) 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.