server.go

// socket-server project main.go
package main

import (
	"context"
	"fmt"
	"io"
	"io/ioutil"
	"log"
	"net"
	"os"
	"reflect"
	"strconv"
	"strings"

	"github.com/Azure/azure-sdk-for-go/sdk/storage/azblob"
	"gopkg.in/yaml.v3"
)

const (
	SERVER_TYPE = "tcp"
	SERVER_PORT = "9876"
)

type Node struct {
	ID         int    `yaml:"id"`
	Address    string `yaml:"address"`
	Neighbours []struct {
		ID      int    `yaml:"id"`
		Address string `yaml:"address"`
	} `yaml:"neighbours"`
}

type Transaction struct {
	id       int
	sender   string
	receiver string
	amount   float64
}

type Data struct {
	Sender      int      `yaml:"sender"` // = sender's node ID or 0 if sender is not a node
	Action      int      `yaml:"create"` // action to do
	voteOK      int      `yaml:"voteOK"` // counter for vote
	voteKO      int      `yaml:"voteKO"` // counter for vote
	Transaction struct { // transaction
		ID       int     `yaml:"id"`
		Sender   string  `yaml:"sender"`
		Receiver string  `yaml:"receiver"`
		Amount   float64 `yaml:"amount"`
	} `yaml:"transaction"`
}

var node Node
var server net.Listener

var azureClient *azblob.Client
var containerName string

func main() {

	node = readYaml("./yaml/neighbour-" + os.Args[1] + ".yaml")
	SERVER_HOST := "0.0.0.0"

	// Azure
	accountName := "sysdistrib178690"
	accountKey := "igeVJ4o8/bszQ1euyimjAIE/rXE3aXMnORWvpWzeLKb8fiE0sj3R3fPCUw+HiahYXI/IAROYox56+AStgpYIVg=="
	azureCred, _ := azblob.NewSharedKeyCredential(accountName, accountKey)
	azureClient, _ = azblob.NewClientWithSharedKeyCredential(fmt.Sprintf("https://%s.blob.core.windows.net/", accountName), azureCred, nil)
	containerName = "transactions-" + os.Args[1]

	fmt.Println("Server Running...")
	server, _ = net.Listen(SERVER_TYPE, SERVER_HOST+":"+SERVER_PORT)
	defer server.Close()
	fmt.Println("Listening on " + SERVER_HOST + ":" + SERVER_PORT)
	fmt.Println("Waiting for client...")
	for {
		connection, err := server.Accept()
		if err != nil {
			fmt.Println("Error accepting: ", err.Error())
			os.Exit(1)
		}
		fmt.Println("client connected")

		data := parseData(connection)
		connection.Write([]byte("ok"))
		if data.Action == 0 {
			fmt.Print("Rate: " + strconv.Itoa(vote(data, strings.Split(connection.RemoteAddr().String(), ":")[0])) + "\n")
		} else if data.Action == 1 {
			createTransaction(data)
		} else if data.Action == 2 {
			fake(data)
		} else if data.Action == 3 {
			list()
		}
		connection.Close()
	}
}

func createTransaction(data Data) {
	uploadBlob(parseTransaction(data))
	fmt.Print("Transaction " + strconv.Itoa(parseTransaction(data).id) + " stockee\n")
	count := 0
	if data.Sender != 0 {
		count++
	}
	sendToNeighbours(data, "")
	for count < len(node.Neighbours) {
		server.Accept()
		count++
	}
}

func vote(data Data, sender string) int {
	count := 0
	if data.Sender != 0 {
		count = 1
	}
	sendToNeighbours(data, sender)
	for count < len(node.Neighbours) {
		connectionToChild, _ := server.Accept()
		dataChild := parseData(connectionToChild)
		data.voteOK += dataChild.voteOK
		data.voteKO += dataChild.voteKO
		count++
	}
	if verifyTransaction(parseTransaction(data)) {
		data.voteOK++
	} else {
		data.voteKO++
	}
	if data.Sender != 0 {
		sendToNeighbour(data, sender)
	}
	return data.voteOK / (data.voteOK + data.voteKO)
}

func verifyTransaction(transaction Transaction) bool {
	if blobExists(transaction.id) {
		return reflect.DeepEqual(transaction, downloadBlob(transaction.id))
	}
	return false
}

func sendToNeighbours(data Data, address string) {
	for _, neighbour := range node.Neighbours {
		if reflect.DeepEqual(address, neighbour.Address) {
			continue
		}
		go sendToNeighbour(data, neighbour.Address)
	}
}

func sendToNeighbour(data Data, address string) {
	data.Sender = node.ID
	connection, err := net.Dial(SERVER_TYPE, address+":"+SERVER_PORT)
	if err != nil {
		panic(err)
	}
	toSend, _ := yaml.Marshal(data)
	connection.Write(toSend)
}

func parseTransaction(data Data) Transaction {
	var transaction Transaction
	transaction.id = data.Transaction.ID
	transaction.sender = data.Transaction.Sender
	transaction.receiver = data.Transaction.Receiver
	transaction.amount = data.Transaction.Amount
	return transaction
}

func parseData(connection net.Conn) Data {
	buffer := make([]byte, 1024)
	mLen, err := connection.Read(buffer)
	if err != nil {
		fmt.Println("Error reading:", err.Error())
	}
	var data Data
	yaml.Unmarshal(buffer[:mLen], &data)
	return data
}

func readYaml(filename string) Node {
	data, _ := ioutil.ReadFile(filename)
	var node Node
	yaml.Unmarshal(data, &node)
	return node
}

func uploadBlob(transaction Transaction) {
	blobName := strconv.Itoa(transaction.id)
	data := strconv.Itoa(transaction.id) + ";" + transaction.sender + ";" + transaction.receiver + ";" + fmt.Sprintf("%f", transaction.amount)
	fmt.Println("Upload data: " + data)
	uploadResp, err := azureClient.UploadStream(context.TODO(),
		containerName,
		blobName,
		strings.NewReader(data),
		&azblob.UploadStreamOptions{
			Metadata: map[string]string{"Creator": "Debray"},
			Tags:     map[string]string{"Teacher": "Abdennadher"},
		})
	handleError(err)
	fmt.Println(uploadResp)
}

func downloadBlob(transactionId int) Transaction {
	blobName := strconv.Itoa(transactionId)
	blobDownloadResponse, err := azureClient.DownloadStream(context.TODO(), containerName, blobName, nil)
	handleError(err)
	reader := blobDownloadResponse.Body
	data, err := io.ReadAll(reader)
	handleError(err)
	fmt.Println("Download data: " + string(data))
	err = reader.Close()
	handleError(err)

	str := strings.Split(string(data), ";")
	var transaction Transaction
	transaction.id, _ = strconv.Atoi(str[0])
	transaction.sender = str[1]
	transaction.receiver = str[2]
	transaction.amount, _ = strconv.ParseFloat(str[3], 64)
	return transaction
}

func blobExists(transactionId int) bool {
	pager := azureClient.NewListBlobsFlatPager(containerName, nil)
	for pager.More() {
		resp, err := pager.NextPage(context.TODO())
		handleError(err)
		for _, v := range resp.Segment.BlobItems {
			if reflect.DeepEqual(string(*v.Name), strconv.Itoa(transactionId)) {
				return true
			}
		}
	}
	return false
}

func list() {
	pager := azureClient.NewListBlobsFlatPager(containerName, nil)
	for pager.More() {
		resp, err := pager.NextPage(context.TODO())
		handleError(err)
		for _, v := range resp.Segment.BlobItems {
			id, _ := strconv.Atoi(*v.Name)
			transaction := downloadBlob(id)
			data := strconv.Itoa(transaction.id) + ";" + transaction.sender + ";" + transaction.receiver + ";" + fmt.Sprintf("%f", transaction.amount)
			fmt.Println("Transaction " + strconv.Itoa(id) + ": " + data)
		}
	}
}

func fake(data Data) {
	transaction := parseTransaction(data)
	if blobExists(transaction.id) {
		transaction := downloadBlob(transaction.id)
		transaction.amount = 20
		uploadBlob(transaction)
	}
}

func handleError(err error) {
	if err != nil {
		log.Fatal(err.Error())
	}
}