站长资源脚本专栏

func ResolveTCPAddr(network, address string) (*TCPAddr, error) {
 switch network {
 case "tcp", "tcp4", "tcp6":
 case "": // a hint wildcard for Go 1.0 undocumented behavior
 network = "tcp"
 default:
 return nil, UnknownNetworkError(network)
 }
 addrs, err := DefaultResolver.internetAddrList(context.Background(), network, address)
 if err != nil {
 return nil, err
 }
 return addrs.forResolve(network, address).(*TCPAddr), nil
}

func DialTCP(network string, laddr, raddr *TCPAddr) (*TCPConn, error) {
 switch network {
 case "tcp", "tcp4", "tcp6":
 default:
 return nil, &OpError{Op: "dial", Net: network, Source: laddr.opAddr(), Addr: raddr.opAddr(), Err: UnknownNetworkError(network)}
 }
 if raddr == nil {
 return nil, &OpError{Op: "dial", Net: network, Source: laddr.opAddr(), Addr: nil, Err: errMissingAddress}
 }
 c, err := dialTCP(context.Background(), network, laddr, raddr)
 if err != nil {
 return nil, &OpError{Op: "dial", Net: network, Source: laddr.opAddr(), Addr: raddr.opAddr(), Err: err}
 }
 return c, nil
}

func Dial(network, address string) (Conn, error) {
 var d Dialer
 return d.Dial(network, address)
}

func Open(addr string) (*bufio.ReadWriter, error) {
 conn, err := net.Dial("tcp", addr)
 if err != nil {
 return nil, errors.Wrap(err, "Dialing "+addr+" failed")
 }
 // 将net.Conn对象包装到bufio.ReadWriter中
 return bufio.NewReadWriter(bufio.NewReader(conn), bufio.NewWriter(conn)), nil
}

type Dialer struct {
 Timeout time.Duration
 Deadline time.Time
 LocalAddr Addr
 DualStack bool
 FallbackDelay time.Duration
 KeepAlive time.Duration
 Resolver *Resolver
 Cancel <-chan struct{}
}

type Conn interface {
 Read(b []byte) (n int, err error)
 Write(b []byte) (n int, err error)
 Close() error
 LocalAddr() Addr
 RemoteAddr() Addr
 SetDeadline(t time.Time) error
 SetReadDeadline(t time.Time) error
 SetWriteDeadline(t time.Time) error
}

type Listener interface {
 // Accept waits for and returns the next connection to the listener.
 Accept() (Conn, error)

 // Close closes the listener.
 // Any blocked Accept operations will be unblocked and return errors.
 Close() error

 // Addr returns the listener's network address.
 Addr() Addr
}

func Listen(network, address string) (Listener, error) {
 addrs, err := DefaultResolver.resolveAddrList(context.Background(), "listen", network, address, nil)
 if err != nil {
 return nil, &OpError{Op: "listen", Net: network, Source: nil, Addr: nil, Err: err}
 }
 var l Listener
 switch la := addrs.first(isIPv4).(type) {
 case *TCPAddr:
 l, err = ListenTCP(network, la)
 case *UnixAddr:
 l, err = ListenUnix(network, la)
 default:
 return nil, &OpError{Op: "listen", Net: network, Source: nil, Addr: la, Err: &AddrError{Err: "unexpected address type", Addr: address}}
 }
 if err != nil {
 return nil, err // l is non-nil interface containing nil pointer
 }
 return l, nil
}

package main

import (
 "bufio"
 "encoding/gob"
 "flag"
 "github.com/pkg/errors"
 "io"
 "log"
 "net"
 "strconv"
 "strings"
 "sync"
)

type complexData struct {
 N int
 S string
 M map[string]int
 P []byte
 C *complexData
}

const (
 Port = ":61000"
)

func Open(addr string) (*bufio.ReadWriter, error) {
 log.Println("Dial " + addr)
 conn, err := net.Dial("tcp", addr)

 if err != nil {
  return nil, errors.Wrap(err, "Dialing " + addr + " failed")
 }

 return bufio.NewReadWriter(bufio.NewReader(conn), bufio.NewWriter(conn)), nil
}

type HandleFunc func(*bufio.ReadWriter)

type Endpoint struct {
 listener net.Listener
 handler map[string]HandleFunc
 m sync.RWMutex  // Maps不是线程安全的，因此需要互斥锁来控制访问。
}

func NewEndpoint() *Endpoint {
 return &Endpoint{
  handler: map[string]HandleFunc{},
 }
}

func (e *Endpoint) AddHandleFunc(name string, f HandleFunc) {
 e.m.Lock()
 e.handler[name] = f
 e.m.Unlock()
}

func (e *Endpoint) Listen() error {
 var err error
 e.listener, err = net.Listen("tcp", Port)
 if err != nil {
  return errors.Wrap(err, "Unable to listen on "+e.listener.Addr().String()+"\n")
 }
 log.Println("Listen on", e.listener.Addr().String())
 for {
  log.Println("Accept a connection request.")
  conn, err := e.listener.Accept()
  if err != nil {
   log.Println("Failed accepting a connection request:", err)
   continue
  }
  log.Println("Handle incoming messages.")
  go e.handleMessages(conn)
 }
}

// handleMessages读取连接到第一个换行符。 基于这个字符串，它会调用恰当的HandleFunc。

func (e *Endpoint) handleMessages(conn net.Conn) {
 // 将连接包装到缓冲reader以便于读取
 rw := bufio.NewReadWrite(bufio.NewReader(conn), bufio.NewWriter(conn))
 defer conn.Close()

 // 从连接读取直到遇到EOF. 期望下一次输入是命令名。调用注册的用于该命令的处理器。

 for {
  log.Print("Receive command '")
  cmd, err := rw.ReadString('\n')
  switch {
  case err == io.EOF:
   log.Println("Reached EOF - close this connection.\n ---")
   return
  case err != nil:
   log.Println("\nError reading command. Got: '" + cmd + "'\n", err)
  }

  // 修剪请求字符串中的多余回车和空格- ReadString不会去掉任何换行。

  cmd = strings.Trim(cmd, "\n ")
  log.Println(cmd + "'")

  // 从handler映射中获取恰当的处理器函数, 并调用它。

  e.m.Lock()
  handleCommand, ok := e.handler[cmd]
  e.m.Unlock()

  if !ok {
   log.Println("Command '" + cmd + "' is not registered.")
   return
  }

  handleCommand(rw)
 }
}

func handleStrings(rw *bufio.ReadWriter) {
 log.Print("Receive STRING message:")
 s, err := rw.ReadString('\n')
 if err != nil {
  log.Println("Cannot read from connection.\n", err)
 }

 s = strings.Trim(s, "\n ")
 log.Println(s)

 -, err = rw.WriteString("Thank you.\n")
 if err != nil {
  log.Println("Cannot write to connection.\n", err)
 }

 err = rw.Flush()
 if err != nil {
  log.Println("Flush failed.", err)
 }
}

func handleGob(rw *bufio.ReadWriter) {
 log.Print("Receive GOB data:")
 var data complexData
  
 dec := gob.NewDecoder(rw)
 err := dec.Decode(&data)

 if err != nil {
  log.Println("Error decoding GOB data:", err)
  return
 }

 log.Printf("Outer complexData struct: \n%#v\n", data)
 log.Printf("Inner complexData struct: \n%#v\n", data.C)
}

// 当应用程序使用-connect=ip地址的时候被调用

func client(ip string) error {
 testStruct := complexData{
  N: 23,
  S: "string data",
  M: map[string]int{"one": 1, "two": 2, "three": 3},
  P: []byte("abc"),
  C: &complexData{
   N: 256,
   S: "Recursive structs",
   M: Map[string]int{"01": "10": 2, "11": 3},
  },
 }

 rw, err := Open(ip + Port)
 if err != nil {
  return errors.Wrap(err, "Client: Failed to open connection to " + ip + Port)
 }

 log.Println("Send the string request.")

 n, err := rw.WriteString("STRING\n")
 if err != nil {
  return errors.Wrap(err, "Could not send the STRING request (" + strconv.Itoa(n) + " bytes written)")
 }

 // 发送STRING请求。发送请求名并发送数据。

 log.Println("Send the string request.")

 n, err = rw.WriteString("Additional data.\n")
 if err != nil {
  return errors.Wrap(err, "Could not send additional STRING data (" + strconv.Itoa(n) + " bytes written)")
 }

 log.Println("Flush the buffer.")
 err = rw.Flush()
 if err != nil {
  return errors.Wrap(err, "Flush failed.")
 }

 // 读取响应

 log.Println("Read the reply.")

 response, err := rw.ReadString('\n')
 if err != nil {
  return errors.Wrap(err, "Client: Failed to read the reply: '" + response + "'")
 }

 log.Println("STRING request: got a response:", response)
 
 // 发送GOB请求。 创建一个encoder直接将它转换为rw.Send的请求名。发送GOB

 log.Println("Send a struct as GOB:")
 log.Printf("Outer complexData struct: \n%#v\n", testStruct)
 log.Printf("Inner complexData struct: \n%#v\n", testStruct.C)
 enc := gob.NewDecoder(rw)
 n, err = rw.WriteString("GOB\n")
 if err != nil {
  return errors.Wrap(err, "Could not write GOB data (" + strconv.Itoa(n) + " bytes written)")
 }

 err = enc.Encode(testStruct)
 if err != nil {
  return errors.Wrap(err, "Encode failed for struct: %#v", testStruct)
 }

 err = rw.Flush()
 if err != nil {
  return errors.Wrap(err, "Flush failed.")
 }
 return nil
}

func server() error {
 endpoint := NewEndpoint()
 // 添加处理器函数
 endpoint.AddHandleFunc("STRING", handleStrings)
 endpoint.AddHandleFunc("GOB", handleGOB)
 // 开始监听
 return endpoint.Listen()
}

func main() {
 connect := flag.String("connect", "", "IP address of process to join. If empty, go into the listen mode.")
 flag.Parse()
 // 如果设置了connect标志，进入客户端模式
 if *connect != '' {
  err := client(*connect)
  if err != nil {
   log.Println("Error:", errors.WithStack(err))
  }
  log.Println("Client done.")
  return
 }
 // 否则进入服务端模式
 err := server()
 if err != nil {
  log.Println("Error:", errors.WithStack(err))
 }
 log.Println("Server done.")
}

// 设置日志记录的字段标志
func init() {
 log.SetFlags(log.Lshortfile)
}

go get -d github.com/appliedgo/networking

cd $GOPATH/src/github.com/appliedgo/networking

go run networking.go

go run networking.go -connect localhost