nsqlookupd
import (
"bufio"
"encoding/binary"
"encoding/json"
"fmt"
"io"
"log"
"net"
"os"
"strings"
"sync/atomic"
"time"
"github.com/nsqio/nsq/internal/protocol"
"github.com/nsqio/nsq/internal/version"
)
type LookupProtocolV1 struct { ctx *Context
}
func (p *LookupProtocolV1) IOLoop(conn net.Conn) error { var err error
var line string
client := NewClientV1(conn)
reader := bufio.NewReader(client)
for { line, err = reader.ReadString(‘\n‘)
if err != nil { break
}
line = strings.TrimSpace(line)
params := strings.Split(line, " ")
var response []byte
response, err = p.Exec(client, reader, params)
if err != nil { ctx := ""
if parentErr := err.(protocol.ChildErr).Parent(); parentErr != nil { ctx = " - " + parentErr.Error()
}
p.ctx.nsqlookupd.logf("ERROR: [%s] - %s%s", client, err, ctx)
_, sendErr := protocol.SendResponse(client, []byte(err.Error()))
if sendErr != nil { p.ctx.nsqlookupd.logf("ERROR: [%s] - %s%s", client, sendErr, ctx) break
}
// errors of type FatalClientErr should forceably close the connection
if _, ok := err.(*protocol.FatalClientErr); ok { break
}
continue
}
if response != nil { _, err = protocol.SendResponse(client, response)
if err != nil { break
}
}
}
conn.Close()
p.ctx.nsqlookupd.logf("CLIENT(%s): closing", client) if client.peerInfo != nil { registrations := p.ctx.nsqlookupd.DB.LookupRegistrations(client.peerInfo.id)
for _, r := range registrations { if removed, _ := p.ctx.nsqlookupd.DB.RemoveProducer(r, client.peerInfo.id); removed { p.ctx.nsqlookupd.logf("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, protocol.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 "", "", protocol.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 !protocol.IsValidTopicName(topicName) { return "", "", protocol.NewFatalClientErr(nil, "E_BAD_TOPIC", fmt.Sprintf("%s topic name ‘%s‘ is not valid", command, topicName)) }
if channelName != "" && !protocol.IsValidChannelName(channelName) { return "", "", protocol.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, protocol.NewFatalClientErr(nil, "E_INVALID", "client must IDENTIFY")
}
topic, channel, err := getTopicChan("REGISTER", params) if err != nil { return nil, err
}
if channel != "" { key := Registration{"channel", topic, channel} if p.ctx.nsqlookupd.DB.AddProducer(key, &Producer{peerInfo: client.peerInfo}) { p.ctx.nsqlookupd.logf("DB: client(%s) REGISTER category:%s key:%s subkey:%s", client, "channel", topic, channel)
}
}
key := Registration{"topic", topic, ""} if p.ctx.nsqlookupd.DB.AddProducer(key, &Producer{peerInfo: client.peerInfo}) { p.ctx.nsqlookupd.logf("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, protocol.NewFatalClientErr(nil, "E_INVALID", "client must IDENTIFY")
}
topic, channel, err := getTopicChan("UNREGISTER", params) if err != nil { return nil, err
}
if channel != "" { key := Registration{"channel", topic, channel} removed, left := p.ctx.nsqlookupd.DB.RemoveProducer(key, client.peerInfo.id)
if removed { p.ctx.nsqlookupd.logf("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
if left == 0 && strings.HasSuffix(channel, "#ephemeral") { p.ctx.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
registrations := p.ctx.nsqlookupd.DB.FindRegistrations("channel", topic, "*") for _, r := range registrations { if removed, _ := p.ctx.nsqlookupd.DB.RemoveProducer(r, client.peerInfo.id); removed { p.ctx.nsqlookupd.logf("WARNING: client(%s) unexpected UNREGISTER category:%s key:%s subkey:%s", client, "channel", topic, r.SubKey)
}
}
key := Registration{"topic", topic, ""} if removed, _ := p.ctx.nsqlookupd.DB.RemoveProducer(key, client.peerInfo.id); removed { p.ctx.nsqlookupd.logf
