Merge pull request #16 from pvelx/add_context_to_net_operation

heap optimisation
added context to net operation
refactoring: merge packages
fix race conditions

Author: pvelx

.gitignore

@@ -13,4 +13,4 @@
 
 # Dependency directories (remove the comment below to include it)
 # vendor/
-.idea/
+.idea/

.pre-commit-config.yaml

@@ -0,0 +1,24 @@
+repos:
+    -   repo: https://github.com/pre-commit/pre-commit-hooks
+        rev: v2.3.0
+        hooks:
+            -   id: check-yaml
+            -   id: end-of-file-fixer
+            -   id: trailing-whitespace
+    -   repo: git://github.com/dnephin/pre-commit-golang
+        rev: master
+        hooks:
+            -   id: golangci-lint
+#            -   id: go-fmt
+#            -   id: go-vet
+#            -   id: go-lint
+#            -   id: go-imports
+#            -   id: go-cyclo
+#                args: [-over=15]
+#            -   id: validate-toml
+#            -   id: no-go-testing
+#            -   id: gometalinter
+#            -   id: go-critic
+#            -   id: go-unit-tests
+#            -   id: go-build
+#            -   id: go-mod-tidy

.travis.yml

@@ -12,4 +12,4 @@ script:
   - make test
 
 notifications:
-  email: false
+  email: false

Makefile

@@ -7,5 +7,7 @@ sending_and_confirmation:
 test:
 	GOMAXPROCS=4 go test ./ -v
 	GOMAXPROCS=4 go test ./repository -v
-	go test ./sender_service ./task_manager ./error_service \
-		./prioritized_task_list ./preloader_service ./monitoring_service ./waiting_service -v
+	go test ./sender_service ./task_manager ./error_service ./preloader_service ./monitoring_service ./waiting_service -v
+
+pre-commit:
+	pre-commit run --all-files

README.md

@@ -1,11 +1,11 @@
 ## Trigger Hook - delayed launch of tasks
 
 [![Build Status](https://travis-ci.com/pvelx/triggerhook.svg?branch=master)](https://travis-ci.com/pvelx/triggerhook)
-[![GitHub release](https://img.shields.io/github/release/pvelx/triggerhook.svg?include_prereleases)](https://github.com/pvelx/triggerhook/releases/latest) 
+[![GitHub release](https://img.shields.io/github/release/pvelx/triggerhook.svg?include_prereleases)](https://github.com/pvelx/triggerhook/releases/latest)
 
-Often in projects, there is a need to perform deferred tasks, 
-such as sending email, push, and other tasks specific to the domain area of your application. 
-Difficulties begin when the usual crontab is no longer enough, 
+Often in projects, there is a need to perform deferred tasks,
+such as sending email, push, and other tasks specific to the domain area of your application.
+Difficulties begin when the usual crontab is no longer enough,
 when batch processing is not suitable, when each task unit has its own execution time or it is assigned dynamically.
 To solve this problem, a Trigger Hook was created. You can build a task scheduler based on this library.
 
@@ -25,11 +25,11 @@ Task  | Description
 
 Life cycle tasks:
 - When creating a task, it gets into the database (square block) (red and yellow).
-- Tasks are loaded into memory (triangular block) if their start time is coming soon (red->yellow). 
+- Tasks are loaded into memory (triangular block) if their start time is coming soon (red->yellow).
 This structure is implemented in the form of a prioritized queue (heap).
-- When the task execution time comes, it is sent for execution (yellow->green). 
+- When the task execution time comes, it is sent for execution (yellow->green).
 An intermediate buffer is used before processing to compensate for peak loads.
-- If the task is successfully submitted, it is deleted from the database (green->blue). 
+- If the task is successfully submitted, it is deleted from the database (green->blue).
 An intermediate buffer is used before deletion, also to compensate for peak loads.
 
 
@@ -39,7 +39,7 @@ Metric|Description
 ---|---
 All|Total number of tasks
 Creating rate | Number of created tasks (via the Create method) per unit of time.
-Deleting rate | Number of deleted tasks (via the Delete method) per unit of time. 
+Deleting rate | Number of deleted tasks (via the Delete method) per unit of time.
 Sending rate | The number of processed tasks (via the Consume method) per unit of time.
 Preloaded | The number of tasks preloaded into memory.
 Preloading rate | The number of tasks loaded per unit of time.
@@ -48,14 +48,14 @@ Waiting for confirmation | The number of tasks waiting for confirmation after se
 Confirmation rate | The number of confirmed tasks after sending per unit of time.
 
 ### Demo
-[Use the demo](https://github.com/pvelx/k8s-message-demo)  
+[Use the demo](https://github.com/pvelx/k8s-message-demo)
 [Read the article](https://vlad-pavlenko.medium.com/deferred-tasks-in-a-microservice-architecture-8e7273089ee7)
 
 ### Features
 - Simple API.
 - Performing tasks with second precision.
 - High performance of sending tasks for execution. This is achieved through a simple task storage scheme, indexing, and multithreaded database access.
-- High peak performance. Tasks that will be completed soon are loaded into memory in advance. This is especially important, for example, if several hundred thousand tasks are assigned at one time. 
+- High peak performance. Tasks that will be completed soon are loaded into memory in advance. This is especially important, for example, if several hundred thousand tasks are assigned at one time.
 - The system is durable to failures. Only after the task is completed, the task is deleted from the database. This ensures that the task is sent for execution. The sudden stop of the application will not lead to inconsistent data in the database.
 - It is designed for a micro-service, event-driven architecture. It is easy to implement a fully asynchronous API.
 - The modular structure of the library. You can easily replace any part with your own implementation.
@@ -125,13 +125,13 @@ func main() {
 				Id:       uuid.NewV4().String(),
 				ExecTime: time.Now().Add(time.Minute).Unix(),
 			}
-			if err := tasksDeferredService.Create(&task); err != nil {
+			if err := tasksDeferredService.CreateCtx(context.Background(), &task); err != nil {
 				log.Fatalf("error creating task: %v", err)
 			}
 
 			// Delete each tenth task
 			if i%10 == 0 {
-				if err := tasksDeferredService.Delete(task.Id); err != nil {
+				if err := tasksDeferredService.DeleteCtx(context.Background(), task.Id); err != nil {
 					log.Fatalf("error deleting task: %v", err)
 				}
 			}

cmd/benchmark/creating_and_deleting.go

@@ -1,6 +1,7 @@
 package main
 
 import (
+	"context"
 	"fmt"
 	"log"
 	"math/rand"
@@ -66,7 +67,7 @@ func deleteTasks(tasks <-chan *domain.Task, triggerHookService contracts.Trigger
 			defer wg.Done()
 			for task := range tasks {
 				preparingBar.Add(1)
-				if err := triggerHookService.Delete(task.Id); err != nil {
+				if err := triggerHookService.DeleteCtx(context.Background(), task.Id); err != nil {
 					log.Fatal(err)
 				}
 			}
@@ -100,7 +101,7 @@ func createTasks(
 				task := &domain.Task{
 					ExecTime: time.Now().Add(time.Hour + time.Duration(rand.Intn(dispersion))*time.Second).Unix(),
 				}
-				if err := triggerHookService.Create(task); err != nil {
+				if err := triggerHookService.CreateCtx(context.Background(), task); err != nil {
 					fmt.Println(err)
 				}
 

cmd/benchmark/main.go

@@ -25,7 +25,7 @@ func clear() {
 		Host:     mysqlHost,
 		DbName:   mysqlDbName,
 	})
-	repository.New(conn, "", nil, nil).Up()
+	_ = repository.New(conn, "", nil, nil).Up()
 
 	if _, err := conn.Exec("DELETE FROM task"); err != nil {
 		log.Fatal(err)

cmd/benchmark/sending_and_confirmation.go

@@ -1,6 +1,7 @@
 package main
 
 import (
+	"context"
 	"fmt"
 	"log"
 	"math/rand"
@@ -49,7 +50,7 @@ func upInitialState(taskCount int) {
 					Id:       util.NewId(),
 					ExecTime: time.Unix(now, 0).Add(-time.Duration(rand.Intn(dispersion)) * time.Second).Unix(),
 				}
-				if err := repositoryService.Create(task, false); err != nil {
+				if err := repositoryService.Create(context.Background(), task, false); err != nil {
 					log.Println(err)
 				}
 			}

contracts/contracts.go

@@ -1,38 +1,13 @@
 package contracts
 
 import (
+	"context"
 	"errors"
 	"time"
 
 	"github.com/pvelx/triggerhook/domain"
 )
 
-/*	--------------------------------------------------
-	Prioritized task list
-*/
-
-type PrioritizedTaskListInterface interface {
-	/*
-		Count of tasks in the list
-	*/
-	Len() int
-
-	/*
-		Add a task to the list based on priority
-	*/
-	Add(task domain.Task)
-
-	/*
-		Take the most prioritized task and delete from list
-	*/
-	Take() *domain.Task
-
-	/*
-		Searches for a task and deletes it return true when the task was deleted, false - when was not found
-	*/
-	DeleteIfExist(taskId string) bool
-}
-
 /*	--------------------------------------------------
 	Sender service
 */
@@ -52,10 +27,10 @@ type TaskToSendInterface interface {
 	Task manager
 */
 type TaskManagerInterface interface {
-	Create(task *domain.Task, isTaken bool) error
-	Delete(taskId string) error
-	GetTasksToComplete(preloadingTimeRange time.Duration) (CollectionsInterface, error)
-	ConfirmExecution(task []domain.Task) error
+	Create(ctx context.Context, task *domain.Task, isTaken bool) error
+	Delete(ctx context.Context, taskId string) error
+	GetTasksToComplete(ctx context.Context, preloadingTimeRange time.Duration) (CollectionsInterface, error)
+	ConfirmExecution(ctx context.Context, task []domain.Task) error
 }
 
 var (
@@ -73,15 +48,15 @@ var (
 	Repository
 */
 type RepositoryInterface interface {
-	Create(task domain.Task, isTaken bool) error
-	Delete(tasks []domain.Task) (int64, error)
-	FindBySecToExecTime(preloadingTimeRange time.Duration) (CollectionsInterface, error)
+	Create(ctx context.Context, task domain.Task, isTaken bool) error
+	Delete(ctx context.Context, tasks []domain.Task) (int64, error)
+	FindBySecToExecTime(ctx context.Context, preloadingTimeRange time.Duration) (CollectionsInterface, error)
 	Up() error
 	Count() (int, error)
 }
 
 type CollectionsInterface interface {
-	Next() (tasks []domain.Task, err error)
+	Next(ctx context.Context) (tasks []domain.Task, err error)
 }
 
 var (
@@ -102,7 +77,7 @@ var (
 	Preloading task service
 */
 type PreloadingServiceInterface interface {
-	AddNewTask(task *domain.Task) error
+	AddNewTask(ctx context.Context, task *domain.Task) error
 	GetPreloadedChan() <-chan domain.Task
 	Run()
 }
@@ -111,7 +86,7 @@ type PreloadingServiceInterface interface {
 	Waiting task service
 */
 type WaitingServiceInterface interface {
-	CancelIfExist(taskId string) error
+	CancelIfExist(ctx context.Context, taskId string) error
 	GetReadyToSendChan() chan domain.Task
 	Run()
 }
@@ -224,22 +199,56 @@ type MonitoringInterface interface {
 */
 
 var (
+	/*
+		Number of tasks waiting for confirmation after sending
+	*/
 	WaitingForConfirmation Topic = "waiting_for_confirmation"
-	ConfirmationRate       Topic = "confirmation_rate"
-	Preloaded              Topic = "preloaded"
-	PreloadingRate         Topic = "preloading_rate"
-	WaitingForSending      Topic = "waiting_for_sending"
-	CreatingRate           Topic = "creating_rate"
-	DeletingRate           Topic = "deleting_rate"
-	SendingRate            Topic = "sending_rate"
-	All                    Topic = "all"
+
+	/*
+		The rate of confirmation of the sending task
+	*/
+	ConfirmationRate Topic = "confirmation_rate"
+
+	/*
+		Number of preloaded tasks
+	*/
+	Preloaded Topic = "preloaded"
+
+	/*
+		Speed of preloading
+	*/
+	PreloadingRate Topic = "preloading_rate"
+
+	/*
+		Deprecated
+		Number of tasks waiting for sending
+	*/
+	WaitingForSending Topic = "waiting_for_sending"
+
+	CreatingRate Topic = "creating_rate"
+
+	DeletingRate Topic = "deleting_rate"
+
+	SendingRate Topic = "sending_rate"
+
+	/*
+		Number of all tasks
+	*/
+	All Topic = "all"
 )
 
 type TriggerHookInterface interface {
+
+	// Deprecated
 	Create(task *domain.Task) error
 
+	// Deprecated
 	Delete(taskId string) error
 
+	CreateCtx(ctx context.Context, task *domain.Task) error
+
+	DeleteCtx(ctx context.Context, taskId string) error
+
 	Consume() TaskToSendInterface
 
 	/*