woop/node.go

package main

import (
	"log"
	"math/rand"
	"time"
)

type Node struct {
	ip          int
	leader      bool
	reciveBlock chan string
	sendBlock   chan string
}

type Nodes struct {
	Nodes []*Node
}

func randomInt(min, max int) int {
	return min + rand.Intn(max-min)
}

// Generate a random string of A-Z chars with len = l
func randomString(len int) string {
	bytes := make([]byte, len)
	for i := 0; i < len; i++ {
		bytes[i] = byte(randomInt(97, 122))
	}
	return string(bytes)
}

func (n Node) send(cin <-chan string, id int) {
	for msg := range cin {
		log.Printf("Leader has sent message %s to %d\n", msg, id)
	}
}

func consume(cin <-chan string, id int) {
	for msg := range cin {
		log.Printf("Leader has sent message %s to %d\n", msg, id)
	}
}

func (n Node) receive() {
	log.Printf("Node: %d received message\n", n.ip)
}

func createNode(ip int, isLeader bool) Node {
	n := Node{ip: ip, leader: isLeader}
	return n
}

func pickLeader(i int) Node {
	if i == 0 {
		return createNode(i, true)
	} else {
		return createNode(i, false)
	}
}

func BufferedTxnQueueWithFanOut(ch <-chan string, size int) []chan string { // This needs
	cs := make([]chan string, size)
	for i, _ := range cs {
		// The size of the channels buffer controls how far behind the recievers
		// of the fanOut channels can lag the other channels.
		cs[i] = make(chan string)
	}
	go func() {
		for txs := range ch {
			for _, c := range cs {
				c <- txs
			}
		}
		for _, c := range cs {
			// close all our fanOut channels when the input channel is exhausted.
			close(c)
		}
	}()
	return cs
}

func TxnGenerator(numOfTxns int, lenOfRandomString int) <-chan string {
	out := make(chan string)
	go func() {
		for i := 0; i < numOfTxns; i++ {
			out <- randomString(lenOfRandomString)
			log.Printf("Transaction Number %d\n", i)
			//time.Sleep(2 * time.Second)
		}
		close(out)
	}()
	return out
}

func main() {
	var (
		//isLeader          Node
		numOfTxns         = 1000
		numOfNodes        = 10
		N                 = make([]Node, 10)
		lenOfRandomString = 10
		node_ips          = []int{1, 2, 3, 4, 5, 6, 7, 8, 9, 10}
	)
	for i := range node_ips {
		m := pickLeader(i)
		N[i] = m
		// if m.leader {
		// 	isLeader := m
		// }
	}
	txnqueue := TxnGenerator(numOfTxns, lenOfRandomString)
	Txns := BufferedTxnQueueWithFanOut(txnqueue, numOfNodes)
	for num := range Txns {
		txn := Txns[num]
		go consume(txn, num)
	}
	time.Sleep(60 * time.Second)
}