mtsatellite/cmd/mtredisalize/changetracker.go

package main

import (
	"encoding/json"
	"log"
	"net/http"
	"sync"

	"bytes"

	"bitbucket.org/s_l_teichmann/mtredisalize/common"
)

// Pull up if it _really_ produces too much data.
const quantizationFactor = 1

type quantizedXZ struct {
	X, Z int16
}

type ChangeTracker struct {
	changes map[quantizedXZ]bool
	mutex   sync.Mutex
}

func NewChangeTracker() *ChangeTracker {
	return &ChangeTracker{changes: make(map[quantizedXZ]bool)}
}

func (ct *ChangeTracker) BlockChanged(key []byte) {
	var err error
	var coord common.Coord
	if coord, err = common.DecodeStringBytesToCoord(key); err != nil {
		log.Printf("decoding key failed: %s", err)
		return
	}
	ct.mutex.Lock()
	ct.changes[quantizedXZ{
		X: coord.X / quantizationFactor,
		Z: coord.Z / quantizationFactor}] = true
	ct.mutex.Unlock()
}

func (ct *ChangeTracker) FlushChanges(url string) (err error) {
	var oldChanges map[quantizedXZ]bool
	ct.mutex.Lock()
	if len(ct.changes) > 0 {
		oldChanges = ct.changes
		ct.changes = make(map[quantizedXZ]bool)
	}
	ct.mutex.Unlock()
	if oldChanges == nil {
		return
	}
	go func() {
		changes := make([]quantizedXZ, len(oldChanges))
		i := 0
		for change, _ := range oldChanges {
			changes[i] = change
			i++
		}
		var buf bytes.Buffer
		encoder := json.NewEncoder(&buf)
		if err = encoder.Encode(changes); err != nil {
			log.Printf("WARN: encode changes to JSON failed: %s", err)
			return
		}
		if _, err = http.Post(
			url, "application/json", bytes.NewBuffer(buf.Bytes())); err != nil {
			log.Printf("WARN: posting changes to %s failed: %s", url, err)
		}
	}()
	return
}