<commit_before>package types

import (
	"context"
	"net/http"
	"time"
)

//------------------------------------------------------------------------------

// Cache is a key/value store that can be shared across components and executing
// threads of a Benthos service.
type Cache interface {
	// Get attempts to locate and return a cached value by its key, returns an
	// error if the key does not exist or if the command fails.
	Get(key string) ([]byte, error)

	// Set attempts to set the value of a key, returns an error if the command
	// fails.
	Set(key string, value []byte) error

	// SetMulti attempts to set the value of multiple keys, returns an error if
	// any of the keys fail.
	SetMulti(items map[string][]byte) error

	// Add attempts to set the value of a key only if the key does not already
	// exist, returns an error if the key already exists or if the command
	// fails.
	Add(key string, value []byte) error

	// Delete attempts to remove a key. Returns an error if a failure occurs.
	Delete(key string) error

	Closable
}

// CacheTTLItem contains a value to cache along with an optional TTL.
type CacheTTLItem struct {
	Value []byte
	TTL   *time.Duration
}

// CacheWithTTL is a key/value store that can be shared across components and
// executing threads of a Benthos service.
type CacheWithTTL interface {
	// SetWithTTL attempts to set the value of a key, returns an error if the
	// command fails.
	SetWithTTL(key string, value []byte, ttl *time.Duration) error

	// SetMultiWithTTL attempts to set the value of multiple keys, returns an
	// error if any of the keys fail.
	SetMultiWithTTL(items map[string]CacheTTLItem) error

	// AddWithTTL attempts to set the value of a key only if the key does not
	// already exist, returns an error if the key already exists or if the
	// command fails.
	AddWithTTL(key string, value []byte, ttl *time.Duration) error

	Cache
}

//------------------------------------------------------------------------------

// RateLimit is a strategy for limiting access to a shared resource, this
// strategy can be safely used by components in parallel.
type RateLimit interface {
	// Access the rate limited resource. Returns a duration or an error if the
	// rate limit check fails. The returned duration is either zero (meaning the
	// resource may be accessed) or a reasonable length of time to wait before
	// requesting again.
	Access() (time.Duration, error)

	Closable
}

//------------------------------------------------------------------------------

// Condition reads a message, calculates a condition and returns a boolean.
type Condition interface {
	// Check tests a message against a configured condition.
	Check(msg Message) bool
}

//------------------------------------------------------------------------------

// Processor reads a message, performs some form of data processing to the
// message, and returns either a slice of >= 1 resulting messages or a response
// to return to the message origin.
type Processor interface {
	// ProcessMessage attempts to process a message. Since processing can fail
	// this call returns both a slice of messages in case of success or a
	// response in case of failure. If the slice of messages is empty the
	// response will be returned to the source.
	ProcessMessage(Message) ([]Message, Response)

	Closable
}

//------------------------------------------------------------------------------

// Manager is an interface expected by Benthos components that allows them to
// register their service wide behaviours such as HTTP endpoints and event
// listeners, and obtain service wide shared resources such as caches.
type Manager interface {
	// RegisterEndpoint registers a server wide HTTP endpoint.
	RegisterEndpoint(path, desc string, h http.HandlerFunc)
<commit_msg>Update processor interface docs to describe errors<commit_after>package types

import (
	"context"
	"net/http"
	"time"
)

//------------------------------------------------------------------------------

// Cache is a key/value store that can be shared across components and executing
// threads of a Benthos service.
type Cache interface {
	// Get attempts to locate and return a cached value by its key, returns an
	// error if the key does not exist or if the command fails.
	Get(key string) ([]byte, error)

	// Set attempts to set the value of a key, returns an error if the command
	// fails.
	Set(key string, value []byte) error

	// SetMulti attempts to set the value of multiple keys, returns an error if
	// any of the keys fail.
	SetMulti(items map[string][]byte) error

	// Add attempts to set the value of a key only if the key does not already
	// exist, returns an error if the key already exists or if the command
	// fails.
	Add(key string, value []byte) error

	// Delete attempts to remove a key. Returns an error if a failure occurs.
	Delete(key string) error

	Closable
}

// CacheTTLItem contains a value to cache along with an optional TTL.
type CacheTTLItem struct {
	Value []byte
	TTL   *time.Duration
}

// CacheWithTTL is a key/value store that can be shared across components and
// executing threads of a Benthos service.
type CacheWithTTL interface {
	// SetWithTTL attempts to set the value of a key, returns an error if the
	// command fails.
	SetWithTTL(key string, value []byte, ttl *time.Duration) error

	// SetMultiWithTTL attempts to set the value of multiple keys, returns an
	// error if any of the keys fail.
	SetMultiWithTTL(items map[string]CacheTTLItem) error

	// AddWithTTL attempts to set the value of a key only if the key does not
	// already exist, returns an error if the key already exists or if the
	// command fails.
	AddWithTTL(key string, value []byte, ttl *time.Duration) error

	Cache
}

//------------------------------------------------------------------------------

// RateLimit is a strategy for limiting access to a shared resource, this
// strategy can be safely used by components in parallel.
type RateLimit interface {
	// Access the rate limited resource. Returns a duration or an error if the
	// rate limit check fails. The returned duration is either zero (meaning the
	// resource may be accessed) or a reasonable length of time to wait before
	// requesting again.
	Access() (time.Duration, error)

	Closable
}

//------------------------------------------------------------------------------

// Condition reads a message, calculates a condition and returns a boolean.
type Condition interface {
	// Check tests a message against a configured condition.
	Check(msg Message) bool
}

//------------------------------------------------------------------------------

// Processor reads a message, performs some form of data processing to the
// message, and returns either a slice of >= 1 resulting messages or a response
// to return to the message origin.
type Processor interface {
	// ProcessMessage attempts to process a message. This method returns both a
	// slice of messages or a response indicating whether messages were dropped
	// due to an intermittent error (response.NewError) or were intentionally
	// filtered (response.NewAck).
	//
	// If an error occurs due to the contents of a message being invalid and you
	// wish to expose this as a recoverable fault you can use processor.FlagErr
	// to flag a message as having failed without dropping it.
	//
	// More information about this form of error handling can be found at:
	// https://www.benthos.dev/docs/configuration/error_handling
	ProcessMessage(Message) ([]Message, Response)

	Closable
}

//------------------------------------------------------------------------------

// Manager is an interface expected by Benthos components that allows them to
// register their service wide behaviours such as HTTP endpoints and event
// listeners, and obtain service wide shared resources such as caches.
type Manager interface {
	// RegisterEndpoint registers a server wide HTTP endpoint.
	RegisterEndpoint(path, desc string, h http.HandlerFunc)

{'fraction_non_alphanumeric': 0.15872617853560683, 'fraction_numerical': 0.00025075225677031093, 'mean_word_length': 2.3629848229342327, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 5, 'https://': 1, 'lorem ipsum': 0, 'www.': 1, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6502082', 'n_tokens_mistral': 1955, 'n_tokens_neox': 1913, 'n_words': 1035}