Changes to change system.

2019-06-16 10:32:34 +02:00
parent 13f82399ef
commit 021f67ea4b
19 changed files with 503 additions and 87 deletions
@@ -0,0 +1,133 @@
+package services
+
+import (
+	"context"
+	"git.aiterp.net/rpdata/api/internal/auth"
+	"git.aiterp.net/rpdata/api/internal/notifier"
+	"git.aiterp.net/rpdata/api/models"
+	"git.aiterp.net/rpdata/api/repositories"
+	"log"
+	"sync"
+	"time"
+)
+
+type ChangeService struct {
+	changes repositories.ChangeRepository
+
+	mutex       sync.RWMutex
+	buffer      []models.Change
+	offset      uint64
+	notifier    notifier.Notifier
+	submitQueue chan *models.Change
+	loopStarted bool
+}
+
+func (s *ChangeService) Find(ctx context.Context, id string) (*models.Change, error) {
+	return s.changes.Find(ctx, id)
+}
+
+func (s *ChangeService) List(ctx context.Context, filter models.ChangeFilter) ([]*models.Change, error) {
+	return s.changes.List(ctx, filter)
+}
+
+func (s *ChangeService) Submit(ctx context.Context, model models.ChangeModel, op string, listed bool, keys []models.ChangeKey, objects ...interface{}) {
+	token := auth.TokenFromContext(ctx)
+	if token == nil {
+		panic("no token!")
+	}
+
+	change := &models.Change{
+		Model:  model,
+		Op:     op,
+		Author: token.UserID,
+		Listed: listed,
+		Keys:   keys,
+	}
+
+	for _, obj := range objects {
+		if !change.AddObject(obj) {
+			log.Printf("Cannot add object of type %T to change", obj)
+		}
+	}
+
+	s.mutex.Lock()
+	if !s.loopStarted {
+		s.loopStarted = true
+		s.submitQueue = make(chan *models.Change, 64)
+		go s.loop()
+	}
+	s.mutex.Unlock()
+
+	s.submitQueue <- change
+}
+
+func (s *ChangeService) Subscribe(ctx context.Context, filter models.ChangeFilter) <-chan *models.Change {
+	channel := make(chan *models.Change)
+
+	go func() {
+		defer close(channel)
+
+		s.mutex.RLock()
+		pos := s.offset + uint64(len(s.buffer))
+		slice := make([]models.Change, 32)
+		s.mutex.RUnlock()
+
+		count := 0
+
+		for {
+			s.mutex.RLock()
+			nextPos := s.offset + uint64(len(s.buffer))
+			length := nextPos - pos
+			if length > 0 {
+				index := pos - s.offset
+
+				pos = nextPos
+				copy(slice, s.buffer[index:])
+			}
+			ch := s.notifier.C()
+			s.mutex.RUnlock()
+
+			for _, change := range slice[:length] {
+				if change.PassesFilter(filter) {
+					channel <- &change
+
+					count++
+					if filter.Limit != nil && count == *filter.Limit {
+						return
+					}
+				}
+			}
+
+			select {
+			case <-ch:
+			case <-ctx.Done():
+				return
+			}
+		}
+	}()
+
+	return channel
+}
+
+func (s *ChangeService) loop() {
+	for change := range s.submitQueue {
+		timeout, cancel := context.WithTimeout(context.Background(), time.Second*15)
+
+		change, err := s.changes.Insert(timeout, *change)
+		if err != nil {
+			log.Println("Failed to submit change:")
+		}
+
+		s.mutex.Lock()
+		s.buffer = append(s.buffer, *change)
+		if len(s.buffer) > 16 {
+			copy(s.buffer, s.buffer[8:])
+			s.buffer = s.buffer[:len(s.buffer)-8]
+			s.offset += 8
+		}
+		s.mutex.Unlock()
+		s.notifier.Broadcast()
+
+		cancel()
+	}
+}
@@ -6,7 +6,6 @@ import (
 	"git.aiterp.net/rpdata/api/internal/auth"
 	"git.aiterp.net/rpdata/api/models"
 	"git.aiterp.net/rpdata/api/models/changekeys"
-	"git.aiterp.net/rpdata/api/models/changes"
 	"git.aiterp.net/rpdata/api/repositories"
 	"git.aiterp.net/rpdata/api/services/loaders"
 	"sort"
@@ -14,8 +13,9 @@ import (
 )

 type CharacterService struct {
-	characters repositories.CharacterRepository
-	loader     *loaders.CharacterLoader
+	characters    repositories.CharacterRepository
+	loader        *loaders.CharacterLoader
+	changeService *ChangeService
 }

 // Find uses the loader to find the character by the ID.
@@ -101,8 +101,7 @@ func (s *CharacterService) Create(ctx context.Context, nick, name, shortName, au
 		return nil, err
 	}

-	//TODO: New change submit system
-	go changes.Submit("Character", "add", token.UserID, true, changekeys.Listed(character), character)
+	s.changeService.Submit(ctx, "Character", "add", true, changekeys.Listed(character), character)

 	return character, nil
 }
@@ -130,9 +129,7 @@ func (s *CharacterService) Update(ctx context.Context, id string, name, shortNam
 	s.loader.Clear(character.ID)
 	s.loader.Prime(character.ID, character)

-	//TODO: New change submit system
-	token := auth.TokenFromContext(ctx)
-	go changes.Submit("Character", "edit", token.UserID, true, changekeys.Listed(character), character)
+	s.changeService.Submit(ctx, "Character", "edit", true, changekeys.Listed(character), character)

 	return character, nil
 }
@@ -153,9 +150,7 @@ func (s *CharacterService) AddNick(ctx context.Context, id string, nick string)
 		return nil, err
 	}

-	//TODO: New change submit system
-	token := auth.TokenFromContext(ctx)
-	go changes.Submit("Character", "edit", token.UserID, true, changekeys.Listed(character), character)
+	s.changeService.Submit(ctx, "Character", "edit", true, changekeys.Listed(character), character)

 	return character, nil
 }
@@ -176,9 +171,7 @@ func (s *CharacterService) RemoveNick(ctx context.Context, id string, nick strin
 		return nil, err
 	}

-	//TODO: New change submit system
-	token := auth.TokenFromContext(ctx)
-	go changes.Submit("Character", "edit", token.UserID, true, changekeys.Listed(character), character)
+	s.changeService.Submit(ctx, "Character", "edit", true, changekeys.Listed(character), character)

 	return character, nil
 }
@@ -199,9 +192,7 @@ func (s *CharacterService) Delete(ctx context.Context, id string) (*models.Chara
 		return nil, err
 	}

-	//TODO: New change submit system
-	token := auth.TokenFromContext(ctx)
-	go changes.Submit("Character", "remove", token.UserID, true, changekeys.Listed(character), character)
+	s.changeService.Submit(ctx, "Character", "remove", true, changekeys.Listed(character), character)

 	return character, nil
 }
@@ -9,16 +9,21 @@ import (
 type Bundle struct {
 	Tags       *TagService
 	Characters *CharacterService
+	Changes    *ChangeService
 }

 // NewBundle creates a new bundle.
 func NewBundle(repos *repositories.Bundle) *Bundle {
 	bundle := &Bundle{}

+	bundle.Changes = &ChangeService{
+		changes: repos.Changes,
+	}
 	bundle.Tags = &TagService{tags: repos.Tags}
 	bundle.Characters = &CharacterService{
-		characters: repos.Characters,
-		loader:     loaders.CharacterLoaderFromRepository(repos.Characters),
+		characters:    repos.Characters,
+		loader:        loaders.CharacterLoaderFromRepository(repos.Characters),
+		changeService: bundle.Changes,
 	}

 	return bundle