feat(operation): Introducing operation entity and thread safe operation queue.

operations/def.go

@@ -0,0 +1,115 @@
+//  Copyright 2021, athena-crdt authors.
+//
+//  Licensed under the Apache License, Version 2.0 (the "License");
+//  you may not use this file except in compliance with the License.
+//  You may obtain a copy of the License at
+//
+//  http://www.apache.org/licenses/LICENSE-2.0
+//
+//  Unless required by applicable law or agreed to in writing, software
+//  distributed under the License is distributed on an "AS IS" BASIS,
+//  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+//  See the License for the specific language governing permissions and
+//  limitations under the License.
+
+package operations
+
+import (
+	"sort"
+	"sync"
+
+	"github.com/pkg/errors"
+
+	"github.com/athena-crdt/athena-core/operations/defs"
+	"github.com/athena-crdt/athena-core/operations/lamport"
+)
+
+type (
+	// Mutation is the type of operation
+	Mutation uint8
+
+	// Operation - individual operation entity
+	Operation struct {
+		Id       *lamport.Clock
+		Deps     []*lamport.Clock
+		Cursor   []defs.NodeId
+		Mutation Mutation
+		Value    defs.Node
+	}
+
+	// OpStore is a thread safe Operation queue.
+	OpStore struct {
+		sync.RWMutex
+		Store []*Operation
+	}
+)
+
+const (
+	ASSIGN = Mutation(iota)
+	INSERT
+	DELETE
+	GET
+)
+
+// NewOpStore creates a new operation store.
+func NewOpStore() *OpStore {
+	return &OpStore{
+		RWMutex: sync.RWMutex{},
+		Store:   []*Operation{},
+	}
+}
+
+func (o *Operation) SortDeps() {
+	sort.Slice(o.Deps, func(i, j int) bool {
+		return o.Deps[i].GetTime() <= o.Deps[j].GetTime()
+	})
+}
+
+func (op *OpStore) Push(o *Operation) {
+	op.Lock()
+	defer op.Unlock()
+	op.Store = append(op.Store, o)
+}
+
+func (op *OpStore) SortedPush(o *Operation) {
+	op.Lock()
+	defer op.Unlock()
+	op.Store = append(op.Store, o)
+	sort.Slice(op.Store, func(i, j int) bool {
+		return op.Store[i].Id.GetTime() <= op.Store[j].Id.GetTime()
+	})
+}
+
+func (op *OpStore) IsEmpty() bool {
+	return op.Len() == 0
+}
+
+func (op *OpStore) Len() int {
+	op.RLock()
+	defer op.RUnlock()
+	return len(op.Store)
+}
+
+func (op *OpStore) Pop() (*Operation, error) {
+	op.Lock()
+	defer op.Unlock()
+	if len(op.Store) == 0 {
+		return nil, errors.New("empty queue")
+	}
+
+	elm := op.Store[0]
+	op.Store = op.Store[1:]
+	return elm, nil
+}
+
+func (op *OpStore) Sort() {
+	sort.Slice(op.Store, func(i, j int) bool {
+		return op.Store[i].Id.GetTime() <= op.Store[j].Id.GetTime()
+	})
+}
+
+func (op *OpStore) Serialize() ([]byte, error) {
+	op.RLock()
+	defer op.RUnlock()
+	panic("implement after PR #12")
+}

operations/lamport/clock.go

@@ -14,29 +14,33 @@
 
 package lamport
 
-import (
-	"sync/atomic"
-)
+import "sync/atomic"
 
 type Clock struct {
+	// counter represents current lamport counter value.
 	counter uint64
+	// hostId identifies the host where the operation was generated at first place.
 	hostId  string
 }
 
-func (clock *Clock) IsGreaterThan(obj *Clock) bool {
-	if atomic.LoadUint64(&obj.counter) == atomic.LoadUint64(&clock.counter) {
-		return clock.hostId > obj.hostId
-	} else {
-		return atomic.LoadUint64(&clock.counter) > atomic.LoadUint64(&obj.counter)
+// NewClock returns a pointer to a newly created clock.
+func NewClock(counter uint64, hostId string) *Clock {
+	return &Clock{
+		hostId: hostId,
+		counter: counter,
 	}
 }
 
+func (clock *Clock) IsGreaterThan(obj *Clock) bool {
+	return atomic.LoadUint64(&clock.counter) > atomic.LoadUint64(&obj.counter)
+}
+
 func (clock *Clock) IsLessThan(obj *Clock) bool {
-	if atomic.LoadUint64(&obj.counter) == atomic.LoadUint64(&clock.counter) {
-		return clock.hostId < obj.hostId
-	} else {
-		return atomic.LoadUint64(&clock.counter) < atomic.LoadUint64(&obj.counter)
-	}
+	 return atomic.LoadUint64(&clock.counter) < atomic.LoadUint64(&obj.counter)
+}
+
+func (clock *Clock) IsEqual(obj *Clock) bool {
+	return atomic.LoadUint64(&clock.counter) == atomic.LoadUint64(&obj.counter)
 }
 
 func (clock *Clock) Increment() uint64 {

operations/lamport/clock_test.go

@@ -42,7 +42,7 @@ func TestClock(t *testing.T) {
 	clock.SetTime(20)
 	assert.Equal(uint64(20), clock.GetTime(), "Invalid time")
 	res := clock.IsGreaterThan(&Clock{20, "def"})
-	assert.Equal(true, res, "Greater than but concludes lesser")
+	assert.Equal(false, res, "Greater than but concludes lesser")
 	res = clock.IsGreaterThan(&Clock{25, "efg"})
 	assert.Equal(false, res, "Lesser than but concludes greater")
 	res = clock.IsLessThan(&Clock{10, "def"})

proto/healthservice.proto

@@ -32,7 +32,7 @@ message HealthCheckResponse {
 }
 
 service HealthService {
-  rpc Check(HealthCheckRequest) returns (HealthCheckResponse);
+  rpc Check (HealthCheckRequest) returns (HealthCheckResponse);
 
-  rpc Watch(HealthCheckRequest) returns (stream HealthCheckResponse);
+  rpc Watch (HealthCheckRequest) returns (stream HealthCheckResponse);
 }

proto/operationservice.proto

@@ -38,7 +38,7 @@ message Operation {
   // Dependency array
   repeated Id dependencies = 2;
   // JsonPath cursor
-  string cursor = 3;
+  repeated string cursor = 3;
   // Type of Operation
   Mutation mutation = 4;
 }