zoreide/orchestrator.go

127 lines
2.7 KiB
Go
Raw Normal View History

2023-08-09 09:08:05 -05:00
package main
import (
"fmt"
"log"
2023-08-09 12:38:57 -05:00
"net"
2023-08-09 09:08:05 -05:00
"slices"
"sort"
"strconv"
"time"
)
func (mip *AbstractMulticast) AddUniqueAndSort(hier int64) {
if slices.Contains(mip.HierarchyArray, hier) {
log.Println("Element already in the array:", hier)
} else {
mip.HierarchyArray = append(mip.HierarchyArray, hier)
sort.Slice(mip.HierarchyArray, func(i, j int) bool { return mip.HierarchyArray[i] < mip.HierarchyArray[j] })
log.Println(mip.HierarchyArray)
}
}
func (mip *AbstractMulticast) IsAlpha(hier int64) bool {
return mip.HierarchyArray[0] == hier
}
func (mip *AbstractMulticast) WriteNumberToMulticast(br AbstractBridge) {
defer func() {
if r := recover(); r != nil {
fmt.Println("An error happened in <WriteNumberToMulticast()>, but Zoreide recovered. ")
fmt.Println("Error was: ", r)
}
}()
log.Println("Initiating ticker")
bstNumber := fmt.Sprintf("%d", br.hIerarchyNumber)
for range time.Tick(1 * time.Second) {
_, err := mip.Wconn.Write([]byte(bstNumber))
if err != nil {
log.Println("Cannot write to multicast:" + err.Error())
}
}
}
func (mip *AbstractMulticast) ReadNumberFromMulticast() {
log.Println("Initiating reader")
buffer := make([]byte, mip.MaxDatagramSize)
// Loop forever reading from the socket
for {
_, _, err := mip.Rconn.ReadFromUDP(buffer)
if err != nil {
log.Println("ReadFromUDP failed:", err)
}
BstChannel <- string(buffer)
}
}
func (b *AbstractBridge) HierarchyReLocator(entity AbstractMulticast) {
log.Println("Inizializing Descalator")
for bstNumber := range BstChannel {
brdNumber, err := strconv.ParseInt(bstNumber, 10, 64)
if err != nil {
log.Println("Garbage received on multicast: ", bstNumber)
log.Println("Cannot convert to int64:", err.Error())
continue
} else {
log.Println("Adding received:", brdNumber)
entity.AddUniqueAndSort(brdNumber)
}
2023-08-09 12:38:57 -05:00
brIp, _, err := net.ParseCIDR(b.BridgeIpCIDR)
if err != nil {
log.Println(err.Error())
}
2023-08-09 09:08:05 -05:00
// finished feeding the new number
// if Alpha:
if entity.IsAlpha(b.hIerarchyNumber) {
2023-08-09 12:38:57 -05:00
if isActive(brIp.String()) {
2023-08-09 09:08:05 -05:00
log.Println("Still ALPHA. This is ok.")
} else {
log.Println("I'm the new ALPHA! Get out my path, losers!")
b.configureIpAndBridgeUp()
2023-08-09 12:38:57 -05:00
log.Println("Ip is active: ", isActive(brIp.String()))
2023-08-09 09:08:05 -05:00
}
} else {
log.Println("GULP! There is a bigger one, better descalating")
2023-08-09 12:38:57 -05:00
if isActive(brIp.String()) {
2023-08-09 09:08:05 -05:00
b.removeIPandBridgeInt()
2023-08-09 12:38:57 -05:00
2023-08-09 09:08:05 -05:00
}
}
}
}
func (b *AbstractBridge) WaitAndClean(entity AbstractMulticast) {
log.Println("Inizializing Escalator")
for {
entity.AddUniqueAndSort(b.hIerarchyNumber)
pollTime := len(entity.HierarchyArray) + 1
time.Sleep(time.Duration(pollTime) * time.Second)
// svuotare l'array
entity.HierarchyArray = entity.HierarchyArray[:0]
}
}