zoreide/orchestrator.go

package main

import (
	"fmt"
	"log"
	"net"
	"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)
		}

		brIp, _, err := net.ParseCIDR(b.BridgeIpCIDR)
		if err != nil {
			log.Println(err.Error())
		}

		// finished feeding the new number
		// if Alpha:
		if entity.IsAlpha(b.hIerarchyNumber) {
			if isActive(brIp.String()) {
				log.Println("Still ALPHA. This is ok.")
			} else {
				log.Println("I'm the new ALPHA! Get out my path, losers!")
				b.configureIpAndBridgeUp()
				log.Println("Ip is active: ", isActive(brIp.String()))
			}
		} else {
			log.Println("GULP! There is a bigger one, better descalating")
			if isActive(brIp.String()) {
				b.removeIPandBridgeInt()

			}
		}

	}

}

func (b *AbstractBridge) WaitAndClean(entity AbstractMulticast) {

	log.Println("Inizializing Escalator")

	entity.AddUniqueAndSort(b.hIerarchyNumber)

	brIp, _, err := net.ParseCIDR(b.BridgeIpCIDR)
	if err != nil {
		log.Println(err.Error())
	}

	for {

		pollTime := len(entity.HierarchyArray) + 1
		time.Sleep(time.Duration(pollTime) * time.Second)
		// svuotare l'array e rifare le elezioni
		if !isActive(string(brIp.String())) {
			entity.HierarchyArray = entity.HierarchyArray[:0]
		}
	}
}