Buffer Compact Lib

Buffer Compactor is a tool to allow for buffering for a duration and compacting data on keys. It uses a badgerDB and sortedset in order to coridinate a time-delayed queue that also aggregates updates sharing the same key in a extremely peformant manner.

BufferCompactor is thread safe and be can shared by consumer and producer threads to delay kafka topics or other message like workloads.

BadgerDB can be configured to live completely on RAM or on Disk depending on workloads.

Example Usage:

Getting Started:

Not handling errors to save space in example

package main

import (
	"fmt"
	"time"

	badger "github.com/dgraph-io/badger/v3"
	"github.com/parkerroan/buffercompact"
)

func main() {
	db, _ := badger.Open(badger.DefaultOptions("").WithInMemory(true))
	bufferDuration := 1 * time.Second

	buffcomp, _ := buffercompact.New(db, bufferDuration, buffercompact.WithMaxValueCount(5))

	buffcomp.StoreToQueue(buffercompact.StorageItem{Key: "test1", Value: []byte("testValue1")})
	buffcomp.StoreToQueue(buffercompact.StorageItem{Key: "test1", Value: []byte("testValue2")})

	limit := 10
	items, _ := buffcomp.RetrieveFromQueue(limit)
	fmt.Printf("No Results: %v \n", items)

	time.Sleep(1 * time.Second)

	items, _ = buffcomp.RetrieveFromQueue(limit)
	fmt.Printf("Results After Buffer Duration: %v \n", items)

	//Insert More Records Up To Max
	buffcomp.StoreToQueue(buffercompact.StorageItem{Key: "test1", Value: []byte("testValue1")})
	buffcomp.StoreToQueue(buffercompact.StorageItem{Key: "test2", Value: []byte("testValue2")})
	buffcomp.StoreToQueue(buffercompact.StorageItem{Key: "test3", Value: []byte("testValue3")})
	buffcomp.StoreToQueue(buffercompact.StorageItem{Key: "test4", Value: []byte("testValue4")})
	buffcomp.StoreToQueue(buffercompact.StorageItem{Key: "test5", Value: []byte("testValue5")})
	buffcomp.StoreToQueue(buffercompact.StorageItem{Key: "test6", Value: []byte("testValue6")})

	items, _ = buffcomp.RetrieveFromQueue(limit)
	fmt.Printf("All Results Returned Due To Max Values Limit (Ignored Buffer Duration): %v \n", items)

	//DEDUPE SECTION
	buffcomp.StoreToQueue(buffercompact.StorageItem{Key: "test1", Value: []byte("testValue1"), UniqueID: "unique-instance-identifier-or-content-hash"})
	time.Sleep(2 * time.Second)
	items, _ = buffcomp.RetrieveFromQueue(limit)
	fmt.Printf("The first time it was treated normally here: %v \n", items)

	//Insert same data again in future...
	buffcomp.StoreToQueue(buffercompact.StorageItem{Key: "test1", Value: []byte("testValue1"), UniqueID: "unique-instance-identifier-or-content-hash"})
	time.Sleep(2 * time.Second)
	items, _ = buffcomp.RetrieveFromQueue(limit)
	fmt.Printf("This time it was ignored as it was already seen: %v \n", items)
}