go语言nsq源码解读七 lookup_protocol_v1.go

package nsqlookupd

import (

"bufio"

"encoding/binary"

"encoding/json"

"fmt"

"io"

"log"

"net"

"os"

"strings"

"time"

"github.com/bitly/go-nsq"

"github.com/bitly/nsq/util"

)

type LookupProtocolV1 struct {

context *Context

}

//实现util\Protocol.go中定义的Protocol的接口的IOLoop方法

func (p *LookupProtocolV1) IOLoop(conn net.Conn) error {

var err error

var line string

//在nsqlookupd\client_v1.go中定义了NewClientV1方法

client := NewClientV1(conn)

err = nil

//此处需要注意为何NewReader可以传入client作为参数。

//打开client_v1.go可以看到，其中嵌入了net.Conn，用JAVA的思想可以说，ClientV1是继承自net.Conn的。

//那接下来的问题是：查官方文档http://golang.org/pkg/bufio/#NewReader

//NewReader的参数类型为io.Reader，这和net.Conn也不同啊

//为一探究竟，我们打开go的源码。分别打开go源码下src\pkg\io\io.go和src\pkg\net\net.go

//发现io.Reader是一个接口，其中有一个方法 Read(p []byte) (n int, err error)

//net.Conn也是一个接口，下面有很多方法，其中一个是 Read(b []byte) (n int, err error)

//可以看出，这两个方法的参数是完全一样的。即net.Conn里的方法完全能覆盖io.Reader里定义的方法

//插播一段关于go接口的描述：所谓Go语言式的接口，就是不用显示声明类型T实现了接口I，只要类型T的公开方法完全满足接口I的要求，就可以把类型T的对象用在需要接口I的地方。这种做法的学名叫做Structural Typing

//所以我们这里可以传入client作为参数

reader := bufio.NewReader(client)

for {

//每次读取一行数据

line, err = reader.ReadString('\n')

if err != nil {

break

}

//去掉两边的空格

line = strings.TrimSpace(line)

//将数据用空格分割成数组，根据后面的代码可看出，第一个参数是动作类型，包括四种：PING IDENTIFY REGISTER UNREGISTER

params := strings.Split(line, " ")

//调用LookupProtocolV1的Exec方法

response, err := p.Exec(client, reader, params)

if err != nil {

context := ""

if parentErr := err.(util.ChildErr).Parent(); parentErr != nil {

context = " - " + parentErr.Error()

}

log.Printf("ERROR: [%s] - %s%s", client, err.Error(), context)

//返回错误给客户端，SendResponse方法在util\Protocol.go中定义

_, err = util.SendResponse(client, []byte(err.Error()))

if err != nil {

break

}

// errors of type FatalClientErr should forceably close the connection

if _, ok := err.(*util.FatalClientErr); ok {

break

}

continue

}

//Exec方法返回了响应数据，将响应发送到客户端

if response != nil {

_, err = util.SendResponse(client, response)

//响应发送出错就退出

if err != nil {

break

}

}

}

//如果前面的for循环退出了，则表示程序要退出了，将注册信息都从RegistrationDB中删除

log.Printf("CLIENT(%s): closing", client)

if client.peerInfo != nil {

registrations := p.context.nsqlookupd.DB.LookupRegistrations(client.peerInfo.id)

for _, r := range registrations {

if removed, _ := p.context.nsqlookupd.DB.RemoveProducer(r, client.peerInfo.id); removed {

log.Printf("DB: client(%s) UNREGISTER category:%s key:%s subkey:%s",

client, r.Category, r.Key, r.SubKey)

}

}

}

return err

}

//请求分发，根据每行数据的第一个参数，调用不同的方法

func (p *LookupProtocolV1) Exec(client *ClientV1, reader *bufio.Reader, params []string) ([]byte, error) {

switch params[0] {

case "PING":

return p.PING(client, params)

case "IDENTIFY":

return p.IDENTIFY(client, reader, params[1:])

case "REGISTER":

return p.REGISTER(client, reader, params[1:])

case "UNREGISTER":

return p.UNREGISTER(client, reader, params[1:])

}

//数据不合法，提示出错。

return nil, util.NewFatalClientErr(nil, "E_INVALID", fmt.Sprintf("invalid command %s", params[0]))

}

func getTopicChan(command string, params []string) (string, string, error) {

if len(params) == 0 {

return "", "", util.NewFatalClientErr(nil, "E_INVALID", fmt.Sprintf("%s insufficient number of params", command))

}

topicName := params[0]

var channelName string

if len(params) >= 2 {

channelName = params[1]

}

if !nsq.IsValidTopicName(topicName) {

return "", "", util.NewFatalClientErr(nil, "E_BAD_TOPIC", fmt.Sprintf("%s topic name '%s' is not valid", command, topicName))

}

if channelName != "" && !nsq.IsValidChannelName(channelName) {

return "", "", util.NewFatalClientErr(nil, "E_BAD_CHANNEL", fmt.Sprintf("%s channel name '%s' is not valid", command, channelName))

}

return topicName, channelName, nil

}

func (p *LookupProtocolV1) REGISTER(client *ClientV1, reader *bufio.Reader, params []string) ([]byte, error) {

if client.peerInfo == nil {

return nil, util.NewFatalClientErr(nil, "E_INVALID", "client must IDENTIFY")

}

//调用本文件里的getTopicChan方法,从参数params中取的topic和channel

topic, channel, err := getTopicChan("REGISTER", params)

if err != nil {

return nil, err

}

if channel != "" {

//定义Registration类型的变量，category为channel,Key为topic,SubKey为channel

key := Registration{"channel", topic, channel}

//将client做为一个producer保存在RegistrationDB中

if p.context.nsqlookupd.DB.AddProducer(key, &Producer{peerInfo: client.peerInfo}) {

log.Printf("DB: client(%s) REGISTER category:%s key:%s subkey:%s",

client, "channel", topic, channel)

}

}

//定义Registration类型的变量，category为topic,Key为topic,SubKey为空

key := Registration{"topic", topic, ""}

//使用另一个key又保存了一次，具体这么做的原因还不是很明确，待确定。

if p.context.nsqlookupd.DB.AddProducer(key, &Producer{peerInfo: client.peerInfo}) {

log.Printf("DB: client(%s) REGISTER category:%s key:%s subkey:%s",

client, "topic", topic, "")

}

return []byte("OK"), nil

}

//反注册，做了和注册相反的事情

func (p *LookupProtocolV1) UNREGISTER(client *ClientV1, reader *bufio.Reader, params []string) ([]byte, error) {

if client.peerInfo == nil {

return nil, util.NewFatalClientErr(nil, "E_INVALID", "client must IDENTIFY")

}

topic, channel, err := getTopicChan("UNREGISTER", params)

if err != nil {

return nil, err

}

//params参数中有指定channel

if channel != "" {

key := Registration{"channel", topic, channel}

removed, left := p.context.nsqlookupd.DB.RemoveProducer(key, client.peerInfo.id)

if removed {

log.Printf("DB: client(%s) UNREGISTER category:%s key:%s subkey:%s",

client, "channel", topic, channel)

}

// for ephemeral channels, remove the channel as well if it has no producers

//把没有producers并标注为ephemeral（中文意思为短暂的）的channels也删除了，此句话翻译了作者的注释

//但是看起来有些不太懂，主要原因是目前对channel topic的概念还不清楚，这个也需待后续明了。

//left=0表示这个Registration下面所有的Producer都被删完了

if left == 0 && strings.HasSuffix(channel, "#ephemeral") {

//将Registration也删除掉了

p.context.nsqlookupd.DB.RemoveRegistration(key)

}

} else {

// no channel was specified so this is a topic unregistration

// remove all of the channel registrations...

// normally this shouldn't happen which is why we print a warning message

// if anything is actually removed

//没有指定channel，所以是topic的反注册

//删除掉所有channel的注册

//一般来说，这不应该发生，如果确实有东西被删掉了，是不正常的，所以在下面的LOG里打印了一个warning

registrations := p.context.nsqlookupd.DB.FindRegistrations("channel", topic, "*")

for _, r := range registrations {

if removed, _ := p.context.nsqlookupd.DB.RemoveProducer(r, client.peerInfo.id); removed {

log.Printf("WARNING: client(%s) unexpected UNREGISTER category:%s key:%s subkey:%s",

client, "channel", topic, r.SubKey)

}

}

key := Registration{"topic", topic, ""}

if removed, _ := p.context.nsqlookupd.DB.RemoveProducer(key, client.peerInfo.id); removed {

log.Printf("DB: client(%s) UNREGISTER category:%s key:%s subkey:%s",

client, "topic", topic, "")

}

}

return []byte("OK"), nil

}

/**

*Client与nsqdlookupd连接后，在进行PING REGISTER UNREGISTER操作之前，必须先IDENTIFY,通过IDENTIFY来初始化peerInfo。

*/

func (p *LookupProtocolV1) IDENTIFY(client *ClientV1, reader *bufio.Reader, params []string) ([]byte, error) {

var err error

//不能重复初始化peerInfo，已经初始化时，返回错误。

if client.peerInfo != nil {

return nil, util.NewFatalClientErr(err, "E_INVALID", "cannot IDENTIFY again")

}

//读取数据的长度到bodyLen变量中，用于判断数据包结尾

var bodyLen int32

err = binary.Read(reader, binary.BigEndian, &bodyLen)

if err != nil {

return nil, util.NewFatalClientErr(err, "E_BAD_BODY", "IDENTIFY failed to read body size")

}

//取bodyLen长度的数据。

body := make([]byte, bodyLen)

_, err = io.ReadFull(reader, body)

if err != nil {

return nil, util.NewFatalClientErr(err, "E_BAD_BODY", "IDENTIFY failed to read body")

}

// body is a json structure with producer information

//body是一个json结构的数据

//PeerInfo类型在nsqlookupd\registration_db.go文件中定义

//PeerInfo类型在Producer类型和ClientV1类型中都有使用

//id作为PeerInfo的唯一性标识，将在后续经常使用

peerInfo := PeerInfo{id: client.RemoteAddr().String()}

err = json.Unmarshal(body, &peerInfo)

if err != nil {

return nil, util.NewFatalClientErr(err, "E_BAD_BODY", "IDENTIFY failed to decode JSON body")

}

peerInfo.RemoteAddress = client.RemoteAddr().String()

// require all fields

//校验JSON传来的数据是否完整

if peerInfo.BroadcastAddress == "" || peerInfo.TcpPort == 0 || peerInfo.HttpPort == 0 || peerInfo.Version == "" {

return nil, util.NewFatalClientErr(nil, "E_BAD_BODY", "IDENTIFY missing fields")

}

//修改peerInfo的lastUpdate值为当前时间

peerInfo.lastUpdate = time.Now()

log.Printf("CLIENT(%s): IDENTIFY Address:%s TCP:%d HTTP:%d Version:%s",

client, peerInfo.BroadcastAddress, peerInfo.TcpPort, peerInfo.HttpPort, peerInfo.Version)

//把当前client加入到RegistrationDB的记录里。Registration的category是"client"，Key和SubKey为空

client.peerInfo = &peerInfo

if p.context.nsqlookupd.DB.AddProducer(Registration{"client", "", ""}, &Producer{peerInfo: client.peerInfo}) {

log.Printf("DB: client(%s) REGISTER category:%s key:%s subkey:%s", client, "client", "", "")

}

// build a response

//构建一个响应给client

data := make(map[string]interface{})

//返回nsqlookupd监听的TCP端口

data["tcp_port"] = p.context.nsqlookupd.tcpAddr.Port

//返回nsqlookupd监听的HTTP端口

data["http_port"] = p.context.nsqlookupd.httpAddr.Port

data["version"] = util.BINARY_VERSION

hostname, err := os.Hostname()

if err != nil {

log.Fatalf("ERROR: unable to get hostname %s", err.Error())

}

data["broadcast_address"] = p.context.nsqlookupd.options.BroadcastAddress

data["hostname"] = hostname

//转化为一个JSON字符串

response, err := json.Marshal(data)

if err != nil {

log.Printf("ERROR: marshaling %v", data)

return []byte("OK"), nil

}

return response, nil

}

func (p *LookupProtocolV1) PING(client *ClientV1, params []string) ([]byte, error) {

if client.peerInfo != nil {

// we could get a PING before other commands on the same client connection

now := time.Now()

log.Printf("CLIENT(%s): pinged (last ping %s)", client.peerInfo.id, now.Sub(client.peerInfo.lastUpdate))

//修改client.peerInfo的lastUpdate值为当前时间

client.peerInfo.lastUpdate = now

}

return []byte("OK"), nil

}