00.02

.github/workflows/golangci-lint.yml

@@ -0,0 +1,37 @@
+name: golangci-lint
+on:
+  push:
+    branches:
+      - main
+  pull_request:
+    branches: [ main, feat/*, rc/* ]
+
+permissions:
+  contents: read
+
+jobs:
+  golangci:
+    name: golangci linter
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/setup-go@v3
+        with:
+          go-version: 1.20.7
+      - uses: actions/checkout@v3
+      - name: golangci-lint
+        uses: golangci/golangci-lint-action@v3
+        with:
+          # Required: the version of golangci-lint is required and must be specified without patch version: we always use the latest patch version.
+          version: v1.53.2
+
+          # Optional: working directory, useful for monorepos
+          # working-directory: somedir
+
+          # Optional: golangci-lint command line arguments.
+          args: --timeout 10m0s --max-issues-per-linter 0 --max-same-issues 0 --print-issued-lines
+
+          # Optional: show only new issues if it's a pull request. The default value is `false`.
+          only-new-issues: true
+
+          # Optional: if set to true then the action will use pre-installed Go
+          # skip-go-installation: true

.github/workflows/pr-tests.yml

@@ -0,0 +1,28 @@
+name: Tests
+
+on:
+  pull_request:
+    branches: [ main, rc/*, feat/* ]
+    types: [ opened, ready_for_review ]
+  push:
+  workflow_dispatch:
+
+jobs:
+  test:
+    name: Unit
+    runs-on: ubuntu-latest
+    steps:
+      - name: Set up Go 1.20.7
+        uses: actions/setup-go@v3
+        with:
+          go-version: 1.20.7
+        id: go
+
+      - name: Check out code
+        uses: actions/checkout@v3
+
+      - name: Get dependencies
+        run: |
+          go get -v -t -d ./...
+      - name: Unit tests
+        run: make test

.gitignore

@@ -0,0 +1,23 @@
+# Binaries for programs and plugins
+*.exe
+*.exe~
+*.dll
+*.so
+*.dylib
+
+# Test binary, built with `go test -c`
+*.test
+
+# Output of the go coverage tool, specifically when used with LiteIDE
+*.out
+
+# Dependency directories
+/vendor
+
+# IDE files
+.idea/
+
+# Go workspace file
+go.work
+
+.idea

Makefile

@@ -0,0 +1,15 @@
+test:
+	@echo "  >  Running unit tests"
+	go test -cover -race -coverprofile=coverage.txt -covermode=atomic -v ./...
+
+lint-install:
+ifeq (,$(wildcard test -f bin/golangci-lint))
+	@echo "Installing golint"
+	curl -sSfL https://raw.githubusercontent.com/golangci/golangci-lint/master/install.sh | sh -s
+endif
+
+run-lint:
+	@echo "Running golint"
+	bin/golangci-lint run --max-issues-per-linter 0 --max-same-issues 0 --timeout=2m
+
+lint: lint-install run-lint

README.md

@@ -0,0 +1,3 @@
+# andes-core-go
+
+andes-go common components and data that can be used in other repositories as well

core/accumulator/export_test.go

@@ -0,0 +1,17 @@
+package accumulator
+
+import "time"
+
+const MinimumAllowedTime = minimumAllowedTime
+
+func (ta *timeAccumulator) Data() []interface{} {
+	ta.mut.Lock()
+	data := make([]interface{}, len(ta.data))
+	ta.mut.Unlock()
+
+	return data
+}
+
+func (ta *timeAccumulator) ComputeWaitTime() time.Duration {
+	return ta.computeWaitTime()
+}

core/accumulator/timeAccumulator.go

@@ -0,0 +1,133 @@
+package accumulator
+
+import (
+	"context"
+	"crypto/rand"
+	"encoding/binary"
+	"fmt"
+	"sync"
+	"time"
+
+	"github.com/subrahamanyam341/andes-abc-01/core"
+	"github.com/subrahamanyam341/andes-abc-01/core/check"
+)
+
+var _ core.Accumulator = (*timeAccumulator)(nil)
+
+const minimumAllowedTime = time.Millisecond * 10
+
+// timeAccumulator is a structure that is able to accumulate data and will try to write on the output channel
+// once per provided interval
+type timeAccumulator struct {
+	cancel         func()
+	maxAllowedTime time.Duration
+	maxOffset      time.Duration
+	mut            sync.Mutex
+	data           []interface{}
+	output         chan []interface{}
+	log            core.Logger
+}
+
+// NewTimeAccumulator returns a new accumulator instance
+func NewTimeAccumulator(maxAllowedTime time.Duration, maxOffset time.Duration, logger core.Logger) (*timeAccumulator, error) {
+	if maxAllowedTime < minimumAllowedTime {
+		return nil, fmt.Errorf("%w for maxAllowedTime as minimum allowed time is %v",
+			core.ErrInvalidValue,
+			minimumAllowedTime,
+		)
+	}
+	if maxOffset < 0 {
+		return nil, fmt.Errorf("%w for maxOffset: should not be negative", core.ErrInvalidValue)
+	}
+	if check.IfNil(logger) {
+		return nil, core.ErrNilLogger
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+
+	ta := &timeAccumulator{
+		cancel:         cancel,
+		maxAllowedTime: maxAllowedTime,
+		output:         make(chan []interface{}),
+		maxOffset:      maxOffset,
+		log:            logger,
+	}
+
+	go ta.continuousEviction(ctx)
+
+	return ta, nil
+}
+
+// AddData will append a new data on the queue
+func (ta *timeAccumulator) AddData(data interface{}) {
+	ta.mut.Lock()
+	ta.data = append(ta.data, data)
+	ta.mut.Unlock()
+}
+
+// OutputChannel returns the output channel on which accumulated data will be sent periodically
+func (ta *timeAccumulator) OutputChannel() <-chan []interface{} {
+	return ta.output
+}
+
+// will call do eviction periodically until the context is done
+func (ta *timeAccumulator) continuousEviction(ctx context.Context) {
+	defer func() {
+		close(ta.output)
+	}()
+
+	for {
+		select {
+		case <-time.After(ta.computeWaitTime()):
+			isDone := ta.doEviction(ctx)
+			if isDone {
+				return
+			}
+		case <-ctx.Done():
+			ta.log.Debug("closing timeAccumulator.continuousEviction go routine")
+			return
+		}
+	}
+}
+
+func (ta *timeAccumulator) computeWaitTime() time.Duration {
+	if ta.maxOffset == 0 {
+		return ta.maxAllowedTime
+	}
+
+	randBuff := make([]byte, 4)
+	_, _ = rand.Read(randBuff)
+	randUint64 := binary.BigEndian.Uint32(randBuff)
+	offset := time.Duration(randUint64) % ta.maxOffset
+
+	return ta.maxAllowedTime - offset
+}
+
+// doEviction will do the eviction of all accumulated data
+// if context.Done is triggered during the eviction, the whole operation will be aborted
+func (ta *timeAccumulator) doEviction(ctx context.Context) bool {
+	ta.mut.Lock()
+	tempData := make([]interface{}, len(ta.data))
+	copy(tempData, ta.data)
+	ta.data = nil
+	ta.mut.Unlock()
+
+	select {
+	case ta.output <- tempData:
+		return false
+	case <-ctx.Done():
+		ta.log.Debug("closing timeAccumulator.doEviction go routine")
+		return true
+	}
+}
+
+// Close stops the time accumulator's eviction loop and closes the output chan
+func (ta *timeAccumulator) Close() error {
+	ta.cancel()
+	return nil
+}
+
+// IsInterfaceNil returns true if there is no value under the interface
+func (ta *timeAccumulator) IsInterfaceNil() bool {
+	return ta == nil
+}