diff --git a/src/inquirer/Makefile b/src/inquirer/Makefile new file mode 100644 index 000000000..c809564de --- /dev/null +++ b/src/inquirer/Makefile @@ -0,0 +1,6 @@ +include $(GOROOT)/src/Make.inc + +TARG=inquirer +GOFILES=inquirer.go + +include $(GOROOT)/src/Make.cmd diff --git a/src/inquirer/inquirer.go b/src/inquirer/inquirer.go new file mode 100644 index 000000000..5afe2af3b --- /dev/null +++ b/src/inquirer/inquirer.go @@ -0,0 +1,52 @@ +package main + +import ( + "fmt" + "log" + "net/http" + "net/rpc" + "registration" + //"time" + "errors" +) + +var raterList *registration.RaterList + +func handler(w http.ResponseWriter, r *http.Request) { + fmt.Fprint(w, "
    ") + for addr, _ := range raterList.Clients { + fmt.Fprint(w, fmt.Sprintf("
  1. %s
    2. ", addr)) + } + fmt.Fprint(w, "
") +} + +func callRater() { + var reply float64 + arg := 9.0 + err := errors.New("") //not nil value + for err != nil { + client:= <-raterList.Balancer + err = client.Call("Sumer.Square", arg, &reply) + if err != nil { + log.Print("Got en error from rater... recovering") + } + } + fmt.Println(fmt.Sprintf("Result: %v", reply)) +} + +func testCallRater(){ + for i:= 0; i<10; i++ { + go callRater() + //time.Sleep(1 * time.Second) + } +} + +func main() { + raterList = registration.NewRaterList() + go testCallRater() + rpc.Register(raterList) + rpc.HandleHTTP() + http.HandleFunc("/", handler) + log.Print("The server is listening...") + http.ListenAndServe(":2000", nil) +} diff --git a/src/inquirer/registration/registration.go b/src/inquirer/registration/registration.go new file mode 100644 index 000000000..9b45dff62 --- /dev/null +++ b/src/inquirer/registration/registration.go @@ -0,0 +1,51 @@ +package registration + +import ( + "fmt" + "log" + "net/rpc" + "time" +) + +type RaterList struct { + Clients map[string]*rpc.Client + Balancer chan *rpc.Client +} + +func NewRaterList() *RaterList { + return &RaterList{ + Clients: make(map[string]*rpc.Client), + Balancer: make(chan *rpc.Client), + } +} + +func (rl *RaterList) RegisterRater(clientAddress string, replay *byte) error { + time.Sleep(1 * time.Second) // wait a second for Rater to start serving + client, err := rpc.Dial("tcp", clientAddress) + if err != nil { + log.Panic("Could not connect to client!") + } + rl.Clients[clientAddress] = client + log.Print(fmt.Sprintf("Server %v registered succesfully", clientAddress)) + rl.startBalance() + return nil +} + +func (rl *RaterList) UnRegisterRater(clientAddress string, replay *byte) error { + client := rl.Clients[clientAddress] + client.Close() + delete(rl.Clients, clientAddress) + log.Print(fmt.Sprintf("Server %v unregistered succesfully", clientAddress)) + return nil +} + +func (rl *RaterList) startBalance() { + go func(){ + for { + for addr, client := range rl.Clients { + log.Printf("using server %s:", addr) + rl.Balancer <- client + } + } + }() +} \ No newline at end of file diff --git a/src/rater/Makefile b/src/rater/Makefile new file mode 100644 index 000000000..c809564de --- /dev/null +++ b/src/rater/Makefile @@ -0,0 +1,6 @@ +include $(GOROOT)/src/Make.inc + +TARG=inquirer +GOFILES=inquirer.go + +include $(GOROOT)/src/Make.cmd diff --git a/src/rater/rater.go b/src/rater/rater.go new file mode 100644 index 000000000..8c5dfc115 --- /dev/null +++ b/src/rater/rater.go @@ -0,0 +1,37 @@ +package main + +import ( + "flag" + "log" + "math" + "net" + "net/rpc" + "registration" +) + +var ( + server = flag.String("server", "127.0.0.1:2000", "target host:port") + listen = flag.String("listen", "127.0.0.1:1234", "target host:port") +) + +type Sumer int + +func (t *Sumer) Square(n float64, reply *float64) error { + *reply = math.Sqrt(n) + return nil +} + +func main() { + flag.Parse() + arith := new(Sumer) + rpc.Register(arith) + rpc.HandleHTTP() + go registration.RegisterToServer(server, listen) + //go registration.StopSingnalHandler(server, listen) + addr, err1 := net.ResolveTCPAddr("tcp", *listen) + l, err2 := net.ListenTCP("tcp", addr) + if err1 != nil || err2 != nil { + log.Panic("cannot create listener for specified address ", *listen) + } + rpc.Accept(l) +} diff --git a/src/rater/registration/registration.go b/src/rater/registration/registration.go new file mode 100644 index 000000000..ca31d710a --- /dev/null +++ b/src/rater/registration/registration.go @@ -0,0 +1,48 @@ +package registration + +import ( + "log" + "net/rpc" + "os" + "os/signal" + "syscall" +) + +func StopSingnalHandler(server, listen *string) { + log.Print("Handling stop signals...") + sig := <-signal.Incoming + if usig, ok := sig.(os.UnixSignal); ok { + switch usig { + case syscall.SIGTERM, syscall.SIGINT, syscall.SIGQUIT: + log.Printf("Caught signal %v, unregistering from server\n", usig) + unregisterFromServer(server, listen) + os.Exit(1) + } + } +} + +func unregisterFromServer(server, listen *string) { + client, err := rpc.DialHTTP("tcp", *server) + if err != nil { + log.Panic("Cannot register to server!") + } + var reply byte + log.Print("Unregistering from server ", *server) + client.Call("RaterList.UnRegisterRater", *listen, &reply) + if err := client.Close(); err != nil { + log.Panic("Could not close server unregistration!") + } +} + +func RegisterToServer(server, listen *string) { + client, err := rpc.DialHTTP("tcp", *server) + if err != nil { + log.Panic("Cannot register to server!") + } + var reply byte + log.Print("Registering to server ", *server) + client.Call("RaterList.RegisterRater", *listen, &reply) + if err := client.Close(); err != nil { + log.Panic("Could not close server registration!") + } +} \ No newline at end of file