WebSocket降级策略(二)

// type messageCache struct {
// 	receiver string
// 	message  string
// 	time     time.Time
// }

type cacheValue struct {
	key   string
	value interface{}
	time  time.Time
}

func (s *cache) Set(key string, kvalue interface{}) error {
	s.clearOutTimeCache()
	value := cacheValue{
		key:   key,
		value: kvalue,
		time:  time.Now(),
	}
	s.Cache = append(s.Cache, value)
	return nil
}

package cacheServer

import (
	"github.com/go-redis/redis"
)

type redisManager struct {
	Name   string
	client *redis.Client
}

type RedisServer interface {
	Get(key string) (string, error)
	Set(key string, value string) error
}

func NewClient(name string, addr string, password string, db int) RedisServer {
	client := redis.NewClient(&redis.Options{
		Addr:     addr,
		Password: password,
		DB:       db,
	})
	return &redisManager{
		Name:   name,
		client: client,
	}
}

func (s *redisManager) Get(key string) (string, error) {
	return s.client.Get(key).Result()
}

func (s *redisManager) Set(key string, value string) error {
	return s.client.Set(key, value, 0).Err()
}

func (s *cache) SetLastWriterToRedis() error {
	return s.RedisClient.Set(s.Name, s.LocalIp)
}

func (s *cache) GetLastWriterFromRedis() {
	IP, err := s.RedisClient.Get(s.Name)
	if err == nil {
		if IP != s.LocalIp {
			resp, err := http.Get(IP + "/syncCache")
			if err != nil {
				return
			}
			defer resp.Body.Close()
			body, _ := ioutil.ReadAll(resp.Body)
			var res []cacheValue
			json.Unmarshal([]byte(body), &res)
			s.Cache = res
		}
	}
}

func NewCache(name string, timeout time.Duration, localIp string, RedisClient RedisServer) CacheServer {
	return &cache{
		Name:        name,
		Cache:       make([]cacheValue, 0),
		Timeout:     timeout,
		RedisClient: RedisClient,
		LocalIp:     localIp,
	}
}

func (s *cache) GetAll() ([]cacheValue, error) {
	s.GetLastWriterFromRedis()
	s.clearOutTimeCache()
	return s.Cache, nil
}

func (s *cache) Get(key string) ([]cacheValue, error) {
	s.GetLastWriterFromRedis()
	s.clearOutTimeCache()
	rlt := make([]cacheValue, 0)
	newCache := make([]cacheValue, 0)
	for i := 0; i < len(s.Cache); i++ {
		if s.Cache[i].Key == key {
			rlt = append(rlt, s.Cache[i])
		} else {
			newCache = append(newCache, s.Cache[i])
		}
	}
	s.Cache = newCache
	s.SetLastWriterToRedis()
	return rlt, nil
}

func (s *cache) Set(key string, kvalue interface{}) error {
	s.GetLastWriterFromRedis()
	s.clearOutTimeCache()
	value := cacheValue{
		Key:   key,
		Value: kvalue,
		Time:  time.Now(),
	}
	s.Cache = append(s.Cache, value)
	s.SetLastWriterToRedis()
	return nil
}

package main

import (
	"cacheServer/cacheServer"
	"encoding/json"
	"fmt"
	"math/rand"
	"net/http"
	"strconv"
	"time"
)

func main() {
	RedisClient := cacheServer.NewClient("localhost:8080", "localhost:6379", "", 0)
	MessageCache1 := cacheServer.NewCache("MessageCache", time.Second*5, "http://localhost:8080", RedisClient)
	MessageCache2 := cacheServer.NewCache("MessageCache", time.Second*5, "http://localhost:8081", RedisClient)
	go Start(":8080", MessageCache1)
	go Start(":8081", MessageCache2)
	go MockWebSocketMessage(MessageCache1)
	select {}
}

func MockWebSocketMessage(cache cacheServer.CacheServer) {
	rand.Seed(time.Now().UnixNano())
	var i int = 0
	for {
		time.Sleep(time.Millisecond * time.Duration(rand.Intn(1000)))
		var receiver string
		if i%2 == 0 {
			receiver = "HYC"
		} else {
			receiver = "ZMN"
		}
		message := "send message " + strconv.Itoa(i) + " times"
		if err := cache.Set(receiver, message); err != nil {
			fmt.Printf("cache Set Error:%s \n", err)
		}
		i = i + 1
	}
}

func Start(Port string, cache cacheServer.CacheServer) error {
	mux := http.NewServeMux()
	mux.HandleFunc("/", health)
	mux.HandleFunc("/syncCache", syncCache(cache))
	mux.HandleFunc("/getCache", getCache(cache))
	svr := &http.Server{Addr: Port, Handler: mux}
	err := svr.ListenAndServe()
	return err
}

func health(w http.ResponseWriter, r *http.Request) {
	fmt.Fprintf(w, "server work")
}

func getCache(cache cacheServer.CacheServer) func(w http.ResponseWriter, r *http.Request) {
	return func(w http.ResponseWriter, r *http.Request) {
		key := GetUrlArg(r, "key")
		value, err := cache.Get(key)
		if err != nil {
			fmt.Printf("cache Get Error:%s \n", err)
		}
		fmt.Fprintf(w, "Cache Key:%s,%v \n", key, value)
	}
}

func syncCache(cache cacheServer.CacheServer) func(w http.ResponseWriter, r *http.Request) {
	return func(w http.ResponseWriter, r *http.Request) {
		value, err := cache.GetAll()
		if err != nil {
			fmt.Printf("cache GetAll Error:%s \n", err)
		}
		res, err := json.Marshal(value)
		if err != nil {
			fmt.Printf("cache json Error:%s \n", err)
		}
		w.Write(res)
	}
}

func GetUrlArg(r *http.Request, name string) string {
	var arg string
	values := r.URL.Query()
	arg = values.Get(name)
	return arg
}

func (s *redisManager) Get(key string) (string, error) {
	return "http://localhost:8080", nil
	//return s.client.Get(key).Result()
}

func (s *redisManager) Set(key string, value string) error {
	return nil
	//return s.client.Set(key, value, 0).Err()
}