recorder

Documentation

Overview

Package recorder contains logic to record data into a database. The payload is guaranteed to be json marshallable. Any types that implements the DataRecorder interface can be used in this system.

Important Notes

When the context is cancelled, the recorder should finish its job and return.

Index

• Variables
• func WithBackoff(backoff int) func(Constructor) error
• func WithEndpoint(endpoint string) func(Constructor) error
• func WithIndexName(indexName string) func(Constructor) error
• func WithLogger(log tools.FieldLogger) func(Constructor) error
• func WithName(name string) func(Constructor) error
• func WithTimeout(timeout time.Duration) func(Constructor) error
• type Constructor
• type DataRecorder
• type EndpointNotAvailableError
  • func (e EndpointNotAvailableError) Error() string
• type InvalidEndpointError
  • func (e InvalidEndpointError) Error() string
• type InvalidIndexNameError
  • func (e InvalidIndexNameError) Error() string
• type Job
• type LowBackoffValueError
  • func (e LowBackoffValueError) Error() string
• type LowTimeout
  • func (e LowTimeout) Error() string
• type ParseTimeOutError
  • func (e ParseTimeOutError) Error() string

Examples

• DataRecorder (Record)
• WithEndpoint
• WithLogger
• WithName

Variables

var (
	ErrEmptyName       = fmt.Errorf("name cannot be empty")
	ErrEmptyEndpoint   = fmt.Errorf("endpoint cannot be empty")
	ErrEmptyIndexName  = fmt.Errorf("index_name cannot be empty")
	ErrBackoffExceeded = fmt.Errorf("endpoint gone too long")
	ErrPingNotCalled   = fmt.Errorf("the caller forgot to ask me pinging")
	ErrNillLogger      = fmt.Errorf("nil logger")
)

Recorder related errors. ErrEmptyEndpoint is returned when the given endpoint is empty. ErrBackoffExceeded is returned 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. ErrPingNotCalled is returned if the caller calls the record without pinging.

Functions

func WithBackoff

func WithBackoff(backoff int) func(Constructor) error

WithBackoff sets the backoff of the recorder

func WithEndpoint

func WithEndpoint(endpoint string) func(Constructor) error

WithEndpoint sets the endpoint of the recorder

Example

r := recorder.Recorder{}
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 WithIndexName

func WithIndexName(indexName string) func(Constructor) error

WithIndexName sets the indexName of the recorder

func WithLogger

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

WithLogger sets the log of the recorder

Example

r := recorder.Recorder{}
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 WithName

func WithName(name string) func(Constructor) error

WithName sets the name of the recorder

Example

r := recorder.Recorder{}
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 recorder

Types

type Constructor

type Constructor interface {
	SetLogger(logger tools.FieldLogger)
	SetName(name string)
	SetIndexName(indexName string)
	SetEndpoint(endpoint string)
	SetTimeout(timeout time.Duration)
	SetBackoff(backoff int)
}

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

type DataRecorder

type DataRecorder interface {
	Name() string
	Record(context.Context, Job) error
	Ping() error
	IndexName() string
	Timeout() time.Duration
	Endpoint() string
	Backoff() int
}

DataRecorder receives a payload for shipping data to a repository. The repository should have the concept of index/database and type/table abstractions. See ElasticSearch for more information.

Notes

Recorders should not change the index name coming in the payload unless they have a valid reason. The engine might add a date to this index name if the user has specified in the configuration file. 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. IndexName() comes from the configuration, but the engine takes over. Recorder() should record the Job and report the errors. When the context is timed-out or cancelled, the recorder should return with the context's error.

Example (Record)

This example shows when a record job is issued, the recorder hits the endpoint.

package main

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

	"github.com/arsham/expipe/datatype"
	"github.com/arsham/expipe/recorder"
	"github.com/arsham/expipe/recorder/testing"
)

func main() {
	ctx := context.Background()
	receivedPayload := make(chan string)
	pinged := false
	ts := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		if !pinged {
			pinged = true
			return
		}
		receivedPayload <- "I have received the payload!"
	}))
	defer ts.Close()

	rec := testing.GetRecorder(ts.URL)
	rec.Ping()

	fmt.Println("Pinging successful")
	payload := datatype.New([]datatype.DataType{
		datatype.NewStringType("key", "value"),
	})
	job := recorder.Job{
		Payload:   payload,
		IndexName: "my index",
		Time:      time.Now(),
	}

	go func() {
		err := rec.Record(ctx, job)
		if err != nil {
			panic("Wasn't expecting any errors")
		}
	}()
	fmt.Println(<-receivedPayload)
	fmt.Println("No errors reported")

	go rec.Record(ctx, job) // Issuing another job
	fmt.Println(<-receivedPayload)

}

Output:

Pinging successful
I have received the payload!
No errors reported
I have received the payload!

type EndpointNotAvailableError

type EndpointNotAvailableError struct {
	Endpoint string
	Err      error
}

EndpointNotAvailableError is returned when the endpoint is not available.

func (EndpointNotAvailableError) Error

func (e EndpointNotAvailableError) Error() string

type InvalidEndpointError

type InvalidEndpointError string

InvalidEndpointError is returned when the endpoint is not a valid URL.

func (InvalidEndpointError) Error

func (e InvalidEndpointError) Error() string

type InvalidIndexNameError

type InvalidIndexNameError string

InvalidIndexNameError is returned when the index name is invalid.

func (InvalidIndexNameError) Error

func (e InvalidIndexNameError) Error() string

type Job

type Job struct {
	// ID is the job ID generated at the time the payload was generated.
	ID token.ID

	// Payload has a Bytes() method for returning the data.
	// It is guaranteed to be json marshallable.
	Payload datatype.DataContainer

	// Time is the recorded time at the time of fetching data by the readers.
	// You should use this value to fetch the content of the payload
	Time time.Time

	// IndexName might be different than the one is set in the recorder.
	// Engine might decide to change it and you have to use the provided one.
	IndexName string

	// TypeName comes from the configuration of readers.
	TypeName string
}

Job is sent with a context and a payload to be recorded. If the TypeName and IndexName are different than the previous one, the recorder should use the ones engine provides. If any errors occurred, recorders should return the error on Read return value.

type LowBackoffValueError

type LowBackoffValueError int64

LowBackoffValueError is returned when the endpoint is not a valid URL.

func (LowBackoffValueError) Error

func (e LowBackoffValueError) Error() string

type LowTimeout

type LowTimeout time.Duration

LowTimeout is returned when the interval is zero.

func (LowTimeout) Error

func (e LowTimeout) Error() string

type ParseTimeOutError

type ParseTimeOutError struct {
	Timeout string
	Err     error
}

ParseTimeOutError is returned when the timeout cannot be parsed.

func (ParseTimeOutError) Error

func (e ParseTimeOutError) Error() string