acc

README

Accumulator: A Go-based Data Accumulation Mechanism

Overview

Accumulator is a lightweight Go library designed to aggregate, manage, and process data items. Whether you're accumulating data from various sources, processing them at specific intervals, or simply managing a stream of data, this library provides the tools you need.

Features

• Flexible Storage: Choose from built-in storage options like in-memory storage or integrate with your custom storage mechanism.
• Customizable Processing: Define how data items are processed using your custom function.
• Configurable Interval: Set the interval at which accumulated data is processed.
• Scheduled Processing: Optionally start processing data at a specific time.

Installation

To install the package, simply run:

go get github.com/9ssi7/acc

Documentation

Flow Diagrams

Continuous Insert with DB Index Issues

graph TD;
    Start[Start] --> Insert1[Insert Data 1];
    Insert1 --> Insert2[Insert Data 2];
    Insert2 --> Insert3[Insert Data 3];
    Insert3 --> Insert4[Insert Data 4];
    Insert4 --> Insert5[Insert Data 5];
    Insert5 --> IndexBroken[Index Broken];

Using Accumulator for Batch Processing

graph TD;
    Start[Start] --> Accumulate[Accumulate Data];
    Accumulate --> Process[Process Data];
    Process --> Wait[Wait 15 Minutes];
    Wait --> Accumulate;

Usage

Basic Setup

Here's a basic example demonstrating how to set up an accumulator:

package main

import (
	"fmt"
	"time"

	"github.com/9ssi7/acc"
)

func main() {
	config := acc.Config[int]{
		Processor: processFunction,
		// ... other configurations
	}

	accumulator := acc.New(config)
	// Start using the accumulator
}

func processFunction(data []int) {
	// Your logic to process data
	fmt.Println("Processing data:", data)
}

Advanced Configurations

Using StartTime Configuration

To start the accumulator processing at a specific time, you can use the StartTime configuration. Below is an example demonstrating how to set a start time:

func main() {
	config := acc.Config[int]{
		Processor: processFunction,
		StartTime: time.Date(2024, time.December, 31, 23, 59, 0, 0, time.UTC), // Processing will start on New Year's Eve at 23:59
		// ... other configurations
	}

	accumulator := acc.New(config)
	// Start using the accumulator
}

Using Interval Configuration

To configure the interval at which accumulated data is processed, you can set the Interval configuration. Below is an example demonstrating how to set an interval:

func main() {
	config := acc.Config[int]{
		Processor: processFunction,
		Interval:  30 * time.Second, // Data will be processed every 30 seconds
		// ... other configurations
	}

	accumulator := acc.New(config)
	// Start using the accumulator
}

Starting the Accumulator Asynchronously

To start the accumulator asynchronously using a goroutine, you can use the following approach:

func main() {
    config := acc.Config[int]{
        Processor: processFunction,
        Interval:  30 * time.Second, // Data will be processed every 30 seconds
        // ... other configurations
    }

    accumulator := acc.New(config)

    // Start the accumulator in a goroutine
    go func() {
        if err := accumulator.Start(); err != nil {
            fmt.Println("Error starting accumulator:", err)
        }
    }()

    // You can continue with other operations or wait for user input to exit
    // For example:
    // fmt.Println("Press Ctrl+C to exit...")
    // select {}
}

By starting the accumulator in a goroutine, it will run in the background, allowing the main program flow to continue without interruption.

Contributing

We welcome contributions! Please see our Contribution Guidelines for details.

License

This project is licensed under the Apache License. See LICENSE for more details.

Documentation

Overview

Package acc provides an accumulator mechanism to accumulate, process, and manage data.

Package acc provides an accumulator mechanism to accumulate, process, and manage data.

Package acc provides an accumulator mechanism to accumulate, process, and manage data.

Package acc provides an accumulator mechanism to accumulate, process, and manage data.

Index

• type Accumulator
  • func New[T any](cnf Config[T]) Accumulator[T]
• type Adder
• type Config
• type Data
• type DataStorage
• type InMemoryStorage
  • func NewInMemoryStorage[T any]() *InMemoryStorage[T]
  • func (s *InMemoryStorage[T]) Load() ([]Data[T], error)
  • func (s *InMemoryStorage[T]) Save(data []Data[T]) error
• type ProcessorFunc

Types

type Accumulator

type Accumulator[T interface{}] interface {
	// Add adds a new item to the accumulator and returns its ID.
	Add(T) (string, error)
	// Cancel removes an item from the accumulator based on the provided ID.
	Cancel(string) error
	// Start initiates the accumulator to process accumulated data.
	Start() error
}

Accumulator defines the interface for the accumulator mechanism.

func New

func New[T any](cnf Config[T]) Accumulator[T]

New initializes a new accumulator instance based on the provided configuration. If no storage is provided, it defaults to an in-memory storage. If no interval is provided, it defaults to 15 minutes.

type Adder

type Adder[T interface{}] func(T) (string, error)

Adder defines the function signature for adding items to the accumulator.

type Config

type Config[T interface{}] struct {

	// Processor is a function to process accumulated items.
	Processor ProcessorFunc[T]

	// Storage represents the storage mechanism to save and load data items.
	Storage DataStorage[Data[T]]

	// Interval specifies the time interval at which the accumulator processes accumulated data.
	Interval time.Duration

	// StartTime sets the specific time when the accumulator should start processing.
	// If set, the accumulator waits until the specified time to begin processing.
	StartTime time.Time
}

Config defines the configuration structure for the accumulator mechanism. It encapsulates the required parameters to initialize an accumulator.

type Data

type Data[T interface{}] struct {
	Original T
	ID       string
}

Data represents a generic data structure with an original value and an ID.

type DataStorage

type DataStorage[T interface{}] interface {
	// Save stores the provided data items.
	Save(data []T) error
	// Load retrieves the stored data items.
	Load() ([]T, error)
}

DataStorage defines the interface for any storage mechanism to save and load data.

type InMemoryStorage

type InMemoryStorage[T any] struct {
	// contains filtered or unexported fields
}

InMemoryStorage is an in-memory storage implementation for the DataStorage interface.

func NewInMemoryStorage

func NewInMemoryStorage[T any]() *InMemoryStorage[T]

NewInMemoryStorage initializes a new instance of InMemoryStorage.

func (*InMemoryStorage[T]) Load

func (s *InMemoryStorage[T]) Load() ([]Data[T], error)

Load retrieves the stored data items from memory.

func (*InMemoryStorage[T]) Save

func (s *InMemoryStorage[T]) Save(data []Data[T]) error

Save stores the provided data items in memory.

type ProcessorFunc

type ProcessorFunc[T interface{}] func(data []T)

ProcessorFunc defines the function signature for processing data.