package pingo
import (
"bufio"
"bytes"
"flag"
"fmt"
"io"
"math/rand"
"net"
"net/rpc"
"os"
"path"
"path/filepath"
"reflect"
"strings"
"time"
)
// Register a new object this plugin exports. The object must be
// an exported symbol and obey all rules an object in the standard
// "rpc" module has to obey.
//注册的一个对象作为可以导出对象。这个对象必须符合RPC规则
// - exported method of exported type
// - two arguments, both of exported type
// - the second argument is a pointer
// - one return value, of type error
// Register will panic if called after Run.
//如果在运行中 注册对象 就会报错
func Register(obj interface{}) {
if defaultServer.running {
panic("Do not call Register after Run")
}
defaultServer.register(obj)//注册可导出的对象
}
// Run will start all the necessary steps to make the plugin available.
//调用Run函数是必须的来保证插件的可用性
func Run() error {
if !flag.Parsed() {//判断参数是否解析
flag.Parse()//解析当前参数
}
return defaultServer.run()
}
// Internal object for plugin control
type PingoRpc struct{}
// Default constructor for interal object. Do not call manually.
func NewPingoRpc() *PingoRpc {
return &PingoRpc{}
}
// Internal RPC call to shut down a plugin. Do not call manually.
func (s *PingoRpc) Exit(status int, unused *int) error {
os.Exit(status)
return nil
}
type config struct {
proto string
addr string
prefix string
unixdir string
}
func makeConfig() *config {
c := &config{}
flag.StringVar(&c.proto, "pingo:proto", "unix", "Protocol to use: unix or tcp")
flag.StringVar(&c.unixdir, "pingo:unixdir", "", "Alternative directory for unix socket")
flag.StringVar(&c.prefix, "pingo:prefix", "pingo", "Prefix to output lines")
return c
}
type rpcServer struct {
*rpc.Server
secret string
objs []string
conf *config
running bool
}
func newRpcServer() *rpcServer {
rand.Seed(time.Now().UTC().UnixNano())
r := &rpcServer{
Server: rpc.NewServer(),
secret: randstr(64),
objs: make([]string, 0),
conf: makeConfig(), // conf remains fixed after this point
}
r.register(&PingoRpc{})
return r
}
var defaultServer = newRpcServer()
type bufReadWriteCloser struct {
*bufio.Reader
r io.ReadWriteCloser
}
func newBufReadWriteCloser(r io.ReadWriteCloser) *bufReadWriteCloser {
return &bufReadWriteCloser{Reader: bufio.NewReader(r), r: r}
}
func (b *bufReadWriteCloser) Write(data []byte) (int, error) {
return b.r.Write(data)
}
func (b *bufReadWriteCloser) Close() error {
return b.r.Close()
}
func readHeaders(brwc *bufReadWriteCloser) ([]byte, error) {
var buf bytes.Buffer
var headerEnd bool
for {
b, err := brwc.ReadByte()
if err != nil {
return []byte(""), err
}
buf.WriteByte(b)
if b == ‘\n‘ {
if headerEnd {
break
}
headerEnd = true
} else {
headerEnd = false
}
}
return buf.Bytes(), nil
}
func parseHeaders(brwc *bufReadWriteCloser, m map[string]string) error {
headers, err := readHeaders(brwc)
if err != nil {
return err
}
r := bytes.NewReader(headers)
scanner := bufio.NewScanner(r)
for scanner.Scan() {
parts := strings.SplitN(scanner.Text(), ": ", 2)
if parts[0] == "" {
continue
}
m[parts[0]] = parts[1]
}
return nil
}
func (r *rpcServer) authConn(token string) bool {
if token != "" && token == r.secret {
return true
}
return false
}
func (r *rpcServer) serveConn(conn io.ReadWriteCloser, h meta) {
bconn := newBufReadWriteCloser(conn)
defer bconn.Close()
headers := make(map[string]string)
if err := parseHeaders(bconn, headers); err != nil {
h.output("error", err.Error())
return
}
if r.authConn(headers["Auth-Token"]) {
r.Server.ServeConn(bconn)
}
return
}
func (r *rpcServer) register(obj interface{}) {
element := reflect.TypeOf(obj).Elem() //获取对象元素类型
r.objs = append(r.objs, element.Name()) //添加到objs中 获取元素类型名称
r.Server.Register(obj)//调用rpc 的注册方法
}
type connection interface {
addr() string
retries() int
}
type tcp int
func (t *tcp) addr() string {
if *t < 1024 {
// Only use unprivileged ports
*t = 1023
}
*t = *t + 1
return fmt.Sprintf("127.0.0.1:%d", *t)
}
func (t *tcp) retries() int {
return 500
}
type unix string
func (u *unix) addr() string {
name := randstr(8)
if *u != "" {
name = filepath.FromSlash(path.Join(string(*u), name))
}
return name
}
func (u *unix) retries() int {
return 4
}
//
func (r *rpcServer) run() error {
var conn connection
var err error
var listener net.Listener
r.running = true //设置默认运行状态为true
h := meta(r.conf.prefix)//设置自定数据类型 参数值为config 的前缀
h.output("objects", strings.Join(r.objs, ", "))//方法参数 key 为常量 objects 参数 val 值为注册服务对象元素名称 。使用，链接的字符串
//协议判断
switch r.conf.proto {
case "tcp":
conn = new(tcp)
default:
r.conf.proto = "unix"
conn = new(unix)
}
//获取尝试连接500次 但是只要有一次执行成功 立刻返回
for i := 0; i < conn.retries(); i++ {
r.conf.addr = conn.addr()
listener, err = net.Listen(r.conf.proto, r.conf.addr)
if err == nil {
break
}
}
if err != nil {
h.output("fatal", fmt.Sprintf("%s: Could not connect in %d attemps, using %s protocol", errorCodeConnFailed, conn.retries(), r.conf.proto))
return err
}
h.output("auth-token", defaultServer.secret)
h.output("ready", fmt.Sprintf("proto=%s addr=%s", r.conf.proto, r.conf.addr))
for {
var conn net.Conn
conn, err = listener.Accept()
if err != nil {
h.output("fatal", fmt.Sprintf("err-http-serve: %s", err.Error()))
continue
}
go r.serveConn(conn, h)
}
}
时间： 2024-10-02 04:26:35