etcd-io__etcd/file_system/event.go

package fileSystem

import (
	"strings"
	"sync"
	"time"
)

const (
	Get        = "get"
	Create     = "create"
	Update     = "update"
	Delete     = "delete"
	TestAndSet = "testAndSet"
)

type Event struct {
	Action     string         `json:"action"`
	Key        string         `json:"key, omitempty"`
	Dir        bool           `json:"dir,omitempty"`
	PrevValue  string         `json:"prevValue,omitempty"`
	Value      string         `json:"value,omitempty"`
	KVPairs    []KeyValuePair `json:"kvs,omitempty"`
	Expiration *time.Time     `json:"expiration,omitempty"`
	TTL        int64          `json:"ttl,omitempty"` // Time to live in second
	// The command index of the raft machine when the command is executed
	Index uint64 `json:"index"`
	Term  uint64 `json:"term"`
}

// When user list a directory, we add all the node into key-value pair slice
type KeyValuePair struct {
	Key     string         `json:"key, omitempty"`
	Value   string         `json:"value,omitempty"`
	Dir     bool           `json:"dir,omitempty"`
	KVPairs []KeyValuePair `json:"kvs,omitempty"`
}

// interfaces for sort
func (e Event) Len() int {
	return len(e.KVPairs)
}

func (e Event) Less(i, j int) bool {
	return e.KVPairs[i].Key < e.KVPairs[j].Key
}

func (e Event) Swap(i, j int) {
	e.KVPairs[i], e.KVPairs[j] = e.KVPairs[j], e.KVPairs[i]
}

func newEvent(action string, key string, index uint64, term uint64) *Event {
	return &Event{
		Action: action,
		Key:    key,
		Index:  index,
		Term:   term,
	}
}

type eventQueue struct {
	events   []*Event
	size     int
	front    int
	capacity int
}

func (eq *eventQueue) back() int {
	return (eq.front + eq.size - 1 + eq.capacity) % eq.capacity
}

func (eq *eventQueue) insert(e *Event) {

	index := (eq.back() + 1) % eq.capacity

	eq.events[index] = e

	if eq.size == eq.capacity { //dequeue
		eq.front = (index + 1) % eq.capacity
	} else {
		eq.size++
	}

}

type EventHistory struct {
	Queue      eventQueue
	StartIndex uint64
	rwl        sync.RWMutex
}

func newEventHistory(capacity int) *EventHistory {
	return &EventHistory{
		Queue: eventQueue{
			capacity: capacity,
			events:   make([]*Event, capacity),
		},
	}
}

// addEvent function adds event into the eventHistory
func (eh *EventHistory) addEvent(e *Event) {
	eh.rwl.Lock()
	defer eh.rwl.Unlock()

	eh.Queue.insert(e)

	eh.StartIndex = eh.Queue.events[eh.Queue.front].Index
}

// scan function is enumerating events from the index in history and
// stops till the first point where the key has identified prefix
func (eh *EventHistory) scan(prefix string, index uint64) (*Event, error) {
	eh.rwl.RLock()
	defer eh.rwl.RUnlock()

	start := index - eh.StartIndex

	// the index should locate after the event history's StartIndex
	// and before its size

	if start < 0 {
		// TODO: Add error type
		return nil, nil
	}

	if start >= uint64(eh.Queue.size) {
		return nil, nil
	}

	i := int((start + uint64(eh.Queue.front)) % uint64(eh.Queue.capacity))

	for {
		e := eh.Queue.events[i]
		if strings.HasPrefix(e.Key, prefix) {
			return e, nil
		}

		i = (i + 1) % eh.Queue.capacity

		if i == eh.Queue.back() {
			// TODO: Add error type
			return nil, nil
		}
	}

}