std - rpc

README.md

@@ -41,4 +41,5 @@
 35. [zerolog](https://darjun.github.io/2020/04/24/godailylib/zerolog)
 36. [nutsdb](https://darjun.github.io/2020/04/25/godailylib/nutsdb)
 37. [sqlc](https://darjun.github.io/2020/04/28/godailylib/sqlc)
-38. [xorm](https://darjun.github.io/2020/05/07/godailylib/xorm)
+38. [xorm](https://darjun.github.io/2020/05/07/godailylib/xorm)
+39. [rpc](https://darjun.github.io/2020/05/08/godailylib/rpc)

rpc/asynchronous/client.go

@@ -0,0 +1,56 @@
+package main
+
+import (
+	"fmt"
+	"log"
+	"net/rpc"
+	"time"
+)
+
+type Args struct {
+	A, B int
+}
+
+type Quotient struct {
+	Quo, Rem int
+}
+
+func main() {
+	client, err := rpc.DialHTTP("tcp", ":1234")
+	if err != nil {
+		log.Fatal("dialing:", err)
+	}
+
+	args1 := &Args{7, 8}
+	var reply int
+	multiplyReply := client.Go("Arith.Multiply", args1, &reply, nil)
+
+	args2 := &Args{15, 6}
+	var quo Quotient
+	divideReply := client.Go("Arith.Divide", args2, &quo, nil)
+
+	ticker := time.NewTicker(time.Millisecond)
+	defer ticker.Stop()
+
+	var multiplyReplied, divideReplied bool
+	for !multiplyReplied || !divideReplied {
+		select {
+		case replyCall := <-multiplyReply.Done:
+			if err := replyCall.Error; err != nil {
+				fmt.Println("Multiply error:", err)
+			} else {
+				fmt.Printf("Multiply: %d*%d=%d\n", args1.A, args1.B, reply)
+			}
+			multiplyReplied = true
+		case replyCall := <-divideReply.Done:
+			if err := replyCall.Error; err != nil {
+				fmt.Println("Divide error:", err)
+			} else {
+				fmt.Printf("Divide: %d/%d=%d...%d\n", args2.A, args2.B, quo.Quo, quo.Rem)
+			}
+			divideReplied = true
+		case <-ticker.C:
+			fmt.Println("tick")
+		}
+	}
+}

rpc/get-started/client.go

@@ -0,0 +1,38 @@
+package main
+
+import (
+	"fmt"
+	"log"
+	"net/rpc"
+)
+
+type Args struct {
+	A, B int
+}
+
+type Quotient struct {
+	Quo, Rem int
+}
+
+func main() {
+	client, err := rpc.DialHTTP("tcp", ":1234")
+	if err != nil {
+		log.Fatal("dialing:", err)
+	}
+
+	args := &Args{7, 8}
+	var reply int
+	err = client.Call("Arith.Multiply", args, &reply)
+	if err != nil {
+		log.Fatal("Multiply error:", err)
+	}
+	fmt.Printf("Multiply: %d*%d=%d\n", args.A, args.B, reply)
+
+	args = &Args{15, 6}
+	var quo Quotient
+	err = client.Call("Arith.Divide", args, &quo)
+	if err != nil {
+		log.Fatal("Divide error:", err)
+	}
+	fmt.Printf("Divide: %d/%d=%d...%d\n", args.A, args.B, quo.Quo, quo.Rem)
+}

rpc/get-started/server/main.go

@@ -0,0 +1,42 @@
+package main
+
+import (
+	"errors"
+	"log"
+	"net/http"
+	"net/rpc"
+)
+
+type Args struct {
+	A, B int
+}
+
+type Quotient struct {
+	Quo, Rem int
+}
+
+type Arith int
+
+func (t *Arith) Multiply(args *Args, reply *int) error {
+	*reply = args.A * args.B
+	return nil
+}
+
+func (t *Arith) Divide(args *Args, quo *Quotient) error {
+	if args.B == 0 {
+		return errors.New("divide by 0")
+	}
+
+	quo.Quo = args.A / args.B
+	quo.Rem = args.A % args.B
+	return nil
+}
+
+func main() {
+	arith := new(Arith)
+	rpc.Register(arith)
+	rpc.HandleHTTP()
+	if err := http.ListenAndServe(":1234", nil); err != nil {
+		log.Fatal("serve error:", err)
+	}
+}

rpc/name/client.go

@@ -0,0 +1,26 @@
+package main
+
+import (
+	"fmt"
+	"log"
+	"net/rpc"
+)
+
+type Args struct {
+	A, B int
+}
+
+func main() {
+	client, err := rpc.DialHTTP("tcp", ":1234")
+	if err != nil {
+		log.Fatal("dialing:", err)
+	}
+
+	args := &Args{7, 8}
+	var reply int
+	err = client.Call("math.Multiply", args, &reply)
+	if err != nil {
+		log.Fatal("Multiply error:", err)
+	}
+	fmt.Printf("Multiply: %d*%d=%d\n", args.A, args.B, reply)
+}

rpc/name/server/main.go

@@ -0,0 +1,31 @@
+package main
+
+import (
+	"log"
+	"net/http"
+	"net/rpc"
+)
+
+type Args struct {
+	A, B int
+}
+
+type Quotient struct {
+	Quo, Rem int
+}
+
+type Arith int
+
+func (t *Arith) Multiply(args *Args, reply *int) error {
+	*reply = args.A * args.B
+	return nil
+}
+
+func main() {
+	arith := new(Arith)
+	rpc.RegisterName("math", arith)
+	rpc.HandleHTTP()
+	if err := http.ListenAndServe(":1234", nil); err != nil {
+		log.Fatal("serve error:", err)
+	}
+}

rpc/newserver/main.go

@@ -0,0 +1,29 @@
+package main
+
+import (
+	"log"
+	"net/http"
+	"net/rpc"
+)
+
+type Args struct {
+	A, B int
+}
+
+type Arith int
+
+func (t *Arith) Multiply(args *Args, reply *int) error {
+	*reply = args.A * args.B
+	return nil
+}
+
+func main() {
+	arith := new(Arith)
+	server := rpc.NewServer()
+	server.RegisterName("math", arith)
+	server.HandleHTTP(rpc.DefaultRPCPath, rpc.DefaultDebugPath)
+
+	if err := http.ListenAndServe(":1234", nil); err != nil {
+		log.Fatal("serve error:", err)
+	}
+}

rpc/servecodec/client.go

@@ -0,0 +1,66 @@
+package main
+
+import (
+	"bufio"
+	"encoding/json"
+	"fmt"
+	"io"
+	"log"
+	"net"
+	"net/rpc"
+)
+
+type Args struct {
+	A, B int
+}
+
+type JsonClientCodec struct {
+	rwc    io.ReadWriteCloser
+	dec    *json.Decoder
+	enc    *json.Encoder
+	encBuf *bufio.Writer
+}
+
+func NewJsonClientCodec(conn io.ReadWriteCloser) *JsonClientCodec {
+	encBuf := bufio.NewWriter(conn)
+	return &JsonClientCodec{conn, json.NewDecoder(conn), json.NewEncoder(encBuf), encBuf}
+}
+
+func (c *JsonClientCodec) WriteRequest(r *rpc.Request, body interface{}) (err error) {
+	if err = c.enc.Encode(r); err != nil {
+		return
+	}
+	if err = c.enc.Encode(body); err != nil {
+		return
+	}
+	return c.encBuf.Flush()
+}
+
+func (c *JsonClientCodec) ReadResponseHeader(r *rpc.Response) error {
+	return c.dec.Decode(r)
+}
+
+func (c *JsonClientCodec) ReadResponseBody(body interface{}) error {
+	return c.dec.Decode(body)
+}
+
+func (c *JsonClientCodec) Close() error {
+	return c.rwc.Close()
+}
+
+func main() {
+	conn, err := net.Dial("tcp", ":1234")
+	if err != nil {
+		log.Fatal("dial error:", err)
+	}
+
+	client := rpc.NewClientWithCodec(NewJsonClientCodec(conn))
+
+	args := &Args{7, 8}
+	var reply int
+	err = client.Call("Arith.Multiply", args, &reply)
+	if err != nil {
+		log.Fatal("Multiply error:", err)
+	}
+	fmt.Printf("Multiply: %d*%d=%d\n", args.A, args.B, reply)
+}

rpc/servecodec/server/main.go

@@ -0,0 +1,87 @@
+package main
+
+import (
+	"bufio"
+	"encoding/json"
+	"io"
+	"log"
+	"net"
+	"net/rpc"
+)
+
+type Args struct {
+	A, B int
+}
+
+type Arith int
+
+func (t *Arith) Multiply(args *Args, reply *int) error {
+	*reply = args.A * args.B
+	return nil
+}
+
+type JsonServerCodec struct {
+	rwc    io.ReadWriteCloser
+	dec    *json.Decoder
+	enc    *json.Encoder
+	encBuf *bufio.Writer
+	closed bool
+}
+
+func NewJsonServerCodec(conn io.ReadWriteCloser) *JsonServerCodec {
+	buf := bufio.NewWriter(conn)
+	return &JsonServerCodec{conn, json.NewDecoder(conn), json.NewEncoder(buf), buf, false}
+}
+
+func (c *JsonServerCodec) ReadRequestHeader(r *rpc.Request) error {
+	return c.dec.Decode(r)
+}
+
+func (c *JsonServerCodec) ReadRequestBody(body interface{}) error {
+	return c.dec.Decode(body)
+}
+
+func (c *JsonServerCodec) WriteResponse(r *rpc.Response, body interface{}) (err error) {
+	if err = c.enc.Encode(r); err != nil {
+		if c.encBuf.Flush() == nil {
+			log.Println("rpc: json error encoding response:", err)
+			c.Close()
+		}
+		return
+	}
+	if err = c.enc.Encode(body); err != nil {
+		if c.encBuf.Flush() == nil {
+			log.Println("rpc: json error encoding body:", err)
+			c.Close()
+		}
+		return
+	}
+	return c.encBuf.Flush()
+}
+
+func (c *JsonServerCodec) Close() error {
+	if c.closed {
+		return nil
+	}
+	c.closed = true
+	return c.rwc.Close()
+}
+
+func main() {
+	l, err := net.Listen("tcp", ":1234")
+	if err != nil {
+		log.Fatal("listen error:", err)
+	}
+
+	arith := new(Arith)
+	rpc.Register(arith)
+
+	for {
+		conn, err := l.Accept()
+		if err != nil {
+			log.Fatal("accept error:", err)
+		}
+
+		go rpc.ServeCodec(NewJsonServerCodec(conn))
+	}
+}