Documentation
¶
Overview ¶
Package consistency implements transactional log management with `Event` and `EventType` abstractions, and supports file-based persistence using `JsonFileLogger` for reliable data storage.
Index ¶
Examples ¶
Constants ¶
This section is empty.
Variables ¶
This section is empty.
Functions ¶
This section is empty.
Types ¶
type Event ¶
type Event[K, V any] struct { Key K `json:"key"` // The key associated with the event. Value V `json:"value"` // The value associated with the event (only for Put). Sequence uint64 `json:"sequence"` // The sequence number of the event, ensuring order. EventType EventType `json:"event_type"` // The type of event (e.g., Put or Delete). }
Event represents an entry in the transactional log.
type EventType ¶
type EventType byte
EventType represents the type of an event in the transactional log.
type JsonFileLogger ¶
type JsonFileLogger[K, V any] struct { // contains filtered or unexported fields }
JsonFileLogger is a file-based implementation of the Logger interface. It writes events to a JSON-formatted file for persistence.
func NewJsonFileLogger ¶
func NewJsonFileLogger[K, V any](file string) *JsonFileLogger[K, V]
NewJsonFileLogger initializes a new JsonFileLogger for the given file path.
Example ¶
package main
import (
"fmt"
"log"
"os"
"path/filepath"
"github.com/andygeiss/cloud-native-utils/consistency"
)
func main() {
dir, err := os.MkdirTemp("", "consistency")
if err != nil {
log.Fatal(err)
}
defer func() { _ = os.RemoveAll(dir) }()
path := filepath.Join(dir, "events.log")
// Every change is appended to the log, in order.
logger := consistency.NewJsonFileLogger[string, string](path)
logger.WritePut("user-1", "Alice")
logger.WritePut("user-2", "Bob")
logger.WriteDelete("user-1")
// Close flushes what is still pending.
if err := logger.Close(); err != nil {
log.Fatal(err)
}
// Replaying the log rebuilds the state it describes.
replay := consistency.NewJsonFileLogger[string, string](path)
defer func() { _ = replay.Close() }()
events, errs := replay.ReadEvents()
for event := range events {
fmt.Println(event.Sequence, event.Key, event.Value)
}
if err := <-errs; err != nil {
log.Fatal(err)
}
}
Output: 1 user-1 Alice 2 user-2 Bob 3 user-1
func (*JsonFileLogger[K, V]) Close ¶
func (a *JsonFileLogger[K, V]) Close() error
Close shuts down the logger, ensuring all pending events are written.
func (*JsonFileLogger[K, V]) Error ¶
func (a *JsonFileLogger[K, V]) Error() <-chan error
Error returns a read-only channel for retrieving errors.
func (*JsonFileLogger[K, V]) ReadEvents ¶
func (a *JsonFileLogger[K, V]) ReadEvents() (<-chan Event[K, V], <-chan error)
ReadEvents reads events from the log file and returns two channels. The method uses a goroutine to read events asynchronously, allowing the caller to process events and handle errors as they are received.
func (*JsonFileLogger[K, V]) WriteDelete ¶
func (a *JsonFileLogger[K, V]) WriteDelete(key K)
WriteDelete writes a delete event to the log.
func (*JsonFileLogger[K, V]) WritePut ¶
func (a *JsonFileLogger[K, V]) WritePut(key K, value V)
WritePut writes a put event to the log.
type Logger ¶
type Logger[K, V any] interface { // Close closes the logger and ensures all pending events are processed. Close() error // WriteDelete writes a delete event to the log. WriteDelete(key K) // WritePut writes a put event to the log. WritePut(key K, value V) // ReadEvents reads events from the log in a streaming manner. ReadEvents() (<-chan Event[K, V], <-chan error) }
Logger is an interface that defines the operations for a transactional log.
Source Files