diff --git a/sylvilagus/go/Makefile b/sylvilagus/go/Makefile index f6efbf4..93fe5f8 100644 --- a/sylvilagus/go/Makefile +++ b/sylvilagus/go/Makefile @@ -3,7 +3,8 @@ PACKAGES := \ $(REPO_BASE)/sylvilagus-conntest \ $(REPO_BASE)/sylvilagus-ch02-hello-world-consumer \ $(REPO_BASE)/sylvilagus-ch02-hello-world-producer \ - $(REPO_BASE)/sylvilagus-ch04-rpc-server + $(REPO_BASE)/sylvilagus-ch04-rpc-server \ + $(REPO_BASE)/sylvilagus-ch04-rpc-client test: build go test -x -v $(PACKAGES) diff --git a/sylvilagus/go/sylvilagus-ch04-rpc-client/main.go b/sylvilagus/go/sylvilagus-ch04-rpc-client/main.go new file mode 100644 index 0000000..a1b953d --- /dev/null +++ b/sylvilagus/go/sylvilagus-ch04-rpc-client/main.go @@ -0,0 +1,63 @@ +package main + +import ( + "encoding/json" + "log" + "time" +) + +import ( + "github.com/meatballhat/box-o-sand/sylvilagus/go/sylvilagus" + "github.com/streadway/amqp" +) + +func main() { + connection, err := amqp.Dial(sylvilagus.AMQP_URI) + if err != nil { + log.Fatal("Failed to connect!:", err) + } + + defer connection.Close() + + channel, err := sylvilagus.CreateRPCTopology(connection) + if err != nil { + log.Fatal("Failed to build topology!:", err) + } + + qResult, err := channel.QueueDeclare("", false, false, true, false, nil) + if err != nil { + log.Fatal("Failed to build topology!:", err) + } + + msgBody := &sylvilagus.Ping{ + ClientName: "RPC Client 1.0", + Time: time.Now(), + } + + msgBytes, err := json.Marshal(msgBody) + if err != nil { + log.Fatal("Failed to json.Marshal the ping!:", err) + } + + msg := amqp.Publishing{ + ContentType: "application/json", + DeliveryMode: amqp.Persistent, + Timestamp: time.Now(), + Body: msgBytes, + ReplyTo: qResult.Name, + } + + channel.Publish("rpc", "ping", false, false, msg) + log.Println("Sent 'ping' RPC call:", string(msg.Body)) + log.Println("Waiting for reply...") + + pongs, err := channel.Consume(qResult.Name, qResult.Name, false, false, false, false, nil) + for pong := range pongs { + log.Println("RPC Reply ---", string(pong.Body)) + if err = channel.Close(); err != nil { + log.Fatal("Failed to close channel!:", err) + } else { + return + } + } +} diff --git a/sylvilagus/go/sylvilagus-ch04-rpc-server/main.go b/sylvilagus/go/sylvilagus-ch04-rpc-server/main.go index 4b32cb0..069d05c 100644 --- a/sylvilagus/go/sylvilagus-ch04-rpc-server/main.go +++ b/sylvilagus/go/sylvilagus-ch04-rpc-server/main.go @@ -1,14 +1,63 @@ package main import ( + "encoding/json" + "fmt" "log" + "time" ) import ( "github.com/meatballhat/box-o-sand/sylvilagus/go/sylvilagus" + "github.com/streadway/amqp" ) func main() { - log.Println("Would be using AMQP_URI =", sylvilagus.AMQP_URI) - log.Println("OH HAI RPC") + connection, err := amqp.Dial(sylvilagus.AMQP_URI) + if err != nil { + log.Fatal("Failed to connect!:", err) + } + + defer connection.Close() + + channel, err := sylvilagus.CreateRPCTopology(connection) + if err != nil { + log.Fatal("Failed to build topology!:", err) + } + + pings, err := channel.Consume("ping", "ping", false, false, false, false, nil) + if err != nil { + log.Fatal("Failed to start consuming!:", err) + } + + quit := make(chan bool) + + go func(quit chan bool) { + log.Println("Waiting for RPC calls...") + + for ping := range pings { + if err = ping.Ack(false); err == nil { + log.Println("Received API call... replying...") + + pingInst := &sylvilagus.Ping{} + err = json.Unmarshal(ping.Body, pingInst) + if err == nil { + msg := amqp.Publishing{ + DeliveryMode: amqp.Persistent, + Timestamp: time.Now(), + ContentType: "text/plain", + Body: []byte(fmt.Sprintf("Pong! %s", pingInst.Time)), + } + + channel.Publish("", ping.ReplyTo, false, false, msg) + } else { + log.Println("Failed to json.Unmarshal the ping!:", err) + } + } else { + log.Println("Failed to ACK the ping!:", err) + } + } + }(quit) + + <-quit } diff --git a/sylvilagus/go/sylvilagus/rpc.go b/sylvilagus/go/sylvilagus/rpc.go index 0b9bb64..c14f6ae 100644 --- a/sylvilagus/go/sylvilagus/rpc.go +++ b/sylvilagus/go/sylvilagus/rpc.go @@ -2,12 +2,18 @@ package sylvilagus import ( "log" + "time" ) import ( "github.com/streadway/amqp" ) +type Ping struct { + ClientName string `json:"client_name"` + Time time.Time `json:"time"` +} + func CreateRPCTopology(connection *amqp.Connection) (channel *amqp.Channel, err error) { if channel, err = connection.Channel(); err != nil { log.Println("Failed to get channel!: ", err)