gogrlx/nats-server
1
0
Fork 0
mirror of https://github.com/gogrlx/nats-server.git synced 2026-04-16 19:14:41 -07:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #9 from apcera/cluster

Browse Source 
Initial Cluster Support
This commit is contained in:
Derek Collison
2013-07-30 16:10:47 -07:00
parent 3ef0c2329a d9017c4cbc
commit 320f03d22b
19 changed files with 1570 additions and 168 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
gnatsd.go
View File

@@ -74,13 +74,19 @@ func main() {
}
// Create the server with appropriate options.
s := server.New(server.MergeOptions(fileOpts, &opts))
mOpts := server.MergeOptions(fileOpts, &opts)
s := server.New(mOpts)
// Start up the http server if needed.
if opts.HttpPort != 0 {
if mOpts.HttpPort != 0 {
s.StartHTTPMonitoring()
}
// Start up routing as well if needed.
if mOpts.ClusterPort != 0 {
s.StartRouting()
}
// Profiler
go func() {
log.Println(http.ListenAndServe("localhost:6062", nil))

263
server/client.go
View File

@@ -19,11 +19,17 @@ import (
// The size of the bufio reader/writer on top of the socket.
const defaultBufSize = 32768
const (
CLIENT = iota
ROUTER
)
type client struct {
mu sync.Mutex
typ int
cid uint64
opts clientOpts
conn net.Conn
nc net.Conn
bw *bufio.Writer
srv *Server
subs *hashmap.HashMap
@@ -33,10 +39,18 @@ type client struct {
pout int
parseState
stats
route *route
}
func (c *client) String() string {
return fmt.Sprintf("cid:%d", c.cid)
func (c *client) String() (id string) {
switch c.typ {
case CLIENT:
id = fmt.Sprintf("cid:%d", c.cid)
case ROUTER:
id = fmt.Sprintf("rid:%d", c.cid)
}
return id
}
type subscription struct {
@@ -72,23 +86,48 @@ func clientConnStr(conn net.Conn) interface{} {
return "N/A"
}
func (c *client) initClient() {
c.mu.Lock()
defer c.mu.Unlock()
s := c.srv
c.cid = atomic.AddUint64(&s.gcid, 1)
c.bw = bufio.NewWriterSize(c.nc, defaultBufSize)
c.subs = hashmap.New()
// This is to track pending clients that have data to be flushed
// after we process inbound msgs from our own connection.
c.pcd = make(map[*client]struct{})
if ip, ok := c.nc.(*net.TCPConn); ok {
ip.SetReadBuffer(defaultBufSize)
}
// Set the Ping timer
c.setPingTimer()
// Spin up the read loop.
go c.readLoop()
}
func (c *client) readLoop() {
// Grab the connection off the client, it will be cleared on a close.
// We check for that after the loop, but want to avoid a nil dereference
conn := c.conn
if conn == nil {
nc := c.nc
if nc == nil {
return
}
b := make([]byte, defaultBufSize)
for {
n, err := conn.Read(b)
n, err := nc.Read(b)
if err != nil {
c.closeConnection()
return
}
if err := c.parse(b[:n]); err != nil {
Log(err.Error(), clientConnStr(c.conn), c.cid)
Log(err.Error(), clientConnStr(c.nc), c.cid)
c.sendErr("Parser Error")
c.closeConnection()
return
@@ -97,10 +136,10 @@ func (c *client) readLoop() {
for cp, _ := range c.pcd {
// Flush those in the set
cp.mu.Lock()
if cp.conn != nil {
cp.conn.SetWriteDeadline(time.Now().Add(DEFAULT_FLUSH_DEADLINE))
if cp.nc != nil {
cp.nc.SetWriteDeadline(time.Now().Add(DEFAULT_FLUSH_DEADLINE))
err := cp.bw.Flush()
cp.conn.SetWriteDeadline(time.Time{})
cp.nc.SetWriteDeadline(time.Time{})
if err != nil {
Debugf("Error flushing: %v", err)
cp.mu.Unlock()
@@ -112,14 +151,14 @@ func (c *client) readLoop() {
delete(c.pcd, cp)
}
// Check to see if we got closed, e.g. slow consumer
if c.conn == nil {
if c.nc == nil {
return
}
}
}
func (c *client) traceMsg(msg []byte) {
pm := fmt.Sprintf("Processing msg: %d", c.inMsgs)
pm := fmt.Sprintf("Processing %s msg: %d", c.typeString(), c.inMsgs)
opa := []interface{}{pm, string(c.pa.subject), string(c.pa.reply), string(msg)}
Trace(logStr(opa), fmt.Sprintf("c: %d", c.cid))
}
@@ -142,17 +181,23 @@ func (c *client) processConnect(arg []byte) error {
// so we can just clear it here.
c.clearAuthTimer()
// FIXME, check err
if err := json.Unmarshal(arg, &c.opts); err != nil {
return err
}
// Check for Auth
if c.srv != nil {
// Check for Auth
if ok := c.srv.checkAuth(c); !ok {
c.sendErr("Authorization is Required")
return fmt.Errorf("Authorization Error")
}
}
// Copy over name if router.
if c.typ == ROUTER && c.route != nil {
c.route.remoteId = c.opts.Name
}
if c.opts.Verbose {
c.sendOK()
}
@@ -182,7 +227,7 @@ func (c *client) sendOK() {
func (c *client) processPing() {
c.traceOp("PING", nil)
if c.conn == nil {
if c.nc == nil {
return
}
c.mu.Lock()
@@ -190,7 +235,7 @@ func (c *client) processPing() {
err := c.bw.Flush()
if err != nil {
c.clearConnection()
Debug("Error on Flush", err, clientConnStr(c.conn), c.cid)
Debug("Error on Flush", err, clientConnStr(c.nc), c.cid)
}
c.mu.Unlock()
}
@@ -202,7 +247,52 @@ func (c *client) processPong() {
c.mu.Unlock()
}
const argsLenMax = 3
func (c *client) processMsgArgs(arg []byte) error {
if trace > 0 {
c.traceOp("MSG", arg)
}
// Unroll splitArgs to avoid runtime/heap issues
a := [MAX_MSG_ARGS][]byte{}
args := a[:0]
start := -1
for i, b := range arg {
switch b {
case ' ', '\t', '\r', '\n':
if start >= 0 {
args = append(args, arg[start:i])
start = -1
}
default:
if start < 0 {
start = i
}
}
}
if start >= 0 {
args = append(args, arg[start:])
}
c.pa.subject = args[0]
c.pa.sid = args[1]
switch len(args) {
case 3:
c.pa.reply = nil
c.pa.szb = args[2]
c.pa.size = parseSize(args[2])
case 4:
c.pa.reply = args[2]
c.pa.szb = args[3]
c.pa.size = parseSize(args[3])
default:
return fmt.Errorf("processMsgArgs Parse Error: '%s'", arg)
}
if c.pa.size < 0 {
return fmt.Errorf("processMsgArgs Bad or Missing Size: '%s'", arg)
}
return nil
}
func (c *client) processPub(arg []byte) error {
if trace > 0 {
@@ -210,7 +300,7 @@ func (c *client) processPub(arg []byte) error {
}
// Unroll splitArgs to avoid runtime/heap issues
a := [argsLenMax][]byte{}
a := [MAX_PUB_ARGS][]byte{}
args := a[:0]
start := -1
for i, b := range arg {
@@ -254,7 +344,7 @@ func (c *client) processPub(arg []byte) error {
}
func splitArg(arg []byte) [][]byte {
a := [argsLenMax][]byte{}
a := [MAX_MSG_ARGS][]byte{}
args := a[:0]
start := -1
for i, b := range arg {
@@ -301,12 +391,16 @@ func (c *client) processSub(argo []byte) (err error) {
if c.srv != nil {
err = c.srv.sl.Insert(sub.subject, sub)
}
shouldForward := c.typ != ROUTER && c.srv != nil
c.mu.Unlock()
if err != nil {
c.sendErr("Invalid Subject")
} else if c.opts.Verbose {
c.sendOK()
}
if shouldForward {
c.srv.broadcastSubscribe(sub)
}
return nil
}
@@ -348,10 +442,14 @@ func (c *client) processUnsub(arg []byte) error {
sub.max = 0
}
c.unsubscribe(sub)
if shouldForward := c.typ != ROUTER && c.srv != nil; shouldForward {
c.srv.broadcastUnSubscribe(sub)
}
}
if c.opts.Verbose {
c.sendOK()
}
return nil
}
@@ -367,10 +465,9 @@ func (c *client) msgHeader(mh []byte, sub *subscription) []byte {
return mh
}
// Used to treat map as efficient set
type empty struct{}
var needFlush = empty{}
// Used to treat maps as efficient set
var needFlush = struct{}{}
var routeSeen = struct{}{}
func (c *client) deliverMsg(sub *subscription, mh, msg []byte) {
if sub.client == nil {
@@ -393,7 +490,7 @@ func (c *client) deliverMsg(sub *subscription, mh, msg []byte) {
}
}
if sub.client.conn == nil {
if sub.client.nc == nil {
client.mu.Unlock()
return
}
@@ -411,7 +508,7 @@ func (c *client) deliverMsg(sub *subscription, mh, msg []byte) {
deadlineSet := false
if client.bw.Available() < (len(mh) + len(msg) + len(CR_LF)) {
client.conn.SetWriteDeadline(time.Now().Add(DEFAULT_FLUSH_DEADLINE))
client.nc.SetWriteDeadline(time.Now().Add(DEFAULT_FLUSH_DEADLINE))
deadlineSet = true
}
@@ -433,7 +530,7 @@ func (c *client) deliverMsg(sub *subscription, mh, msg []byte) {
}
if deadlineSet {
client.conn.SetWriteDeadline(time.Time{})
client.nc.SetWriteDeadline(time.Time{})
}
client.mu.Unlock()
@@ -442,13 +539,13 @@ func (c *client) deliverMsg(sub *subscription, mh, msg []byte) {
writeErr:
if deadlineSet {
client.conn.SetWriteDeadline(time.Time{})
client.nc.SetWriteDeadline(time.Time{})
}
client.mu.Unlock()
if ne, ok := err.(net.Error); ok && ne.Timeout() {
// FIXME: SlowConsumer logic
Log("Slow Consumer Detected", clientConnStr(client.conn), client.cid)
Log("Slow Consumer Detected", clientConnStr(client.nc), client.cid)
client.closeConnection()
} else {
Debugf("Error writing msg: %v", err)
@@ -459,9 +556,12 @@ func (c *client) processMsg(msg []byte) {
c.inMsgs++
c.inBytes += int64(len(msg))
if c.srv != nil {
atomic.AddInt64(&c.srv.inMsgs, 1)
atomic.AddInt64(&c.srv.inBytes, int64(len(msg)))
// Snapshot server.
srv := c.srv
if srv != nil {
atomic.AddInt64(&srv.inMsgs, 1)
atomic.AddInt64(&srv.inBytes, int64(len(msg)))
}
if trace > 0 {
@@ -474,8 +574,10 @@ func (c *client) processMsg(msg []byte) {
c.sendOK()
}
scratch := [512]byte{}
msgh := scratch[:0]
// The msg header starts with "MSG ",
// in bytes that is [77 83 71 32].
scratch := [512]byte{77, 83, 71, 32}
msgh := scratch[:4]
r := c.srv.sl.Match(c.pa.subject)
if len(r) <= 0 {
@@ -483,8 +585,6 @@ func (c *client) processMsg(msg []byte) {
}
// msg header
// FIXME, put MSG into initializer
msgh = append(msgh, "MSG "...)
msgh = append(msgh, c.pa.subject...)
msgh = append(msgh, ' ')
si := len(msgh)
@@ -492,14 +592,36 @@ func (c *client) processMsg(msg []byte) {
var qmap map[string][]*subscription
var qsubs []*subscription
isRoute := c.typ == ROUTER
var rmap map[string]struct{}
// If we are a route and we have a queue subscription, deliver direct
// since they are sent direct via L2 semantics.
if isRoute {
if sub := c.srv.routeSidQueueSubscriber(c.pa.sid); sub != nil {
mh := c.msgHeader(msgh[:si], sub)
c.deliverMsg(sub, mh, msg)
return
}
}
// Loop over all subscriptions that match.
for _, v := range r {
sub := v.(*subscription)
// Process queue group subscriptions by gathering them all up
// here. We will pick the winners when we are done processing
// all of the subscriptions.
if sub.queue != nil {
// FIXME, this can be more efficient
// Queue subscriptions handled from routes directly above.
if isRoute {
continue
}
// FIXME(dlc), this can be more efficient
if qmap == nil {
qmap = make(map[string][]*subscription)
}
//qname := *(*string)(unsafe.Pointer(&sub.queue))
qname := string(sub.queue)
qsubs = qmap[qname]
if qsubs == nil {
@@ -509,9 +631,37 @@ func (c *client) processMsg(msg []byte) {
qmap[qname] = qsubs
continue
}
// Process normal, non-queue group subscriptions.
// If this is a send to a ROUTER, make sure we only send it
// once. The other side will handle the appropriate re-processing.
// Also enforce 1-Hop.
if sub.client.typ == ROUTER {
// Skip if sourced from a ROUTER and going to another ROUTER.
// This is 1-Hop semantics for ROUTERs.
if isRoute {
continue
}
// Check to see if we have already sent it here.
if rmap == nil {
rmap = make(map[string]struct{}, len(srv.routes))
}
if sub.client == nil || sub.client.route == nil || sub.client.route.remoteId == "" {
Debug("Bad or Missing ROUTER Identity, not processing msg", clientConnStr(c.nc), c.cid)
continue
}
if _, ok := rmap[sub.client.route.remoteId]; ok {
continue
}
rmap[sub.client.route.remoteId] = routeSeen
}
mh := c.msgHeader(msgh[:si], sub)
c.deliverMsg(sub, mh, msg)
}
if qmap != nil {
for _, qsubs := range qmap {
index := rand.Int() % len(qsubs)
@@ -527,16 +677,16 @@ func (c *client) processPingTimer() {
defer c.mu.Unlock()
c.ptmr = nil
// Check if we are ready yet..
if _, ok := c.conn.(*net.TCPConn); !ok {
if _, ok := c.nc.(*net.TCPConn); !ok {
return
}
Debug("Ping Timer", clientConnStr(c.conn), c.cid)
Debug("Client Ping Timer", clientConnStr(c.nc), c.cid)
// Check for violation
c.pout += 1
if c.pout > c.srv.opts.MaxPingsOut {
Debug("Stale Connection - Closing", clientConnStr(c.conn), c.cid)
Debug("Stale Client Connection - Closing", clientConnStr(c.nc), c.cid)
if c.bw != nil {
c.bw.WriteString(fmt.Sprintf("-ERR '%s'\r\n", "Stale Connection"))
c.bw.Flush()
@@ -549,7 +699,7 @@ func (c *client) processPingTimer() {
c.bw.WriteString("PING\r\n")
err := c.bw.Flush()
if err != nil {
Debug("Error on Flush", err, clientConnStr(c.conn), c.cid)
Debug("Error on Client Ping Flush", err, clientConnStr(c.nc), c.cid)
c.clearConnection()
} else {
// Reset to fire again if all OK.
@@ -596,39 +746,58 @@ func (c *client) isAuthTimerSet() bool {
// Lock should be held
func (c *client) clearConnection() {
if c.conn == nil {
if c.nc == nil {
return
}
c.bw.Flush()
c.conn.Close()
c.conn = nil
c.nc.Close()
c.nc = nil
}
func (c *client) typeString() string {
switch c.typ {
case CLIENT:
return "Client"
case ROUTER:
return "Router"
}
return "Unknown Type"
}
func (c *client) closeConnection() {
c.mu.Lock()
if c.conn == nil {
if c.nc == nil {
c.mu.Unlock()
return
}
Debug("Client connection closed", clientConnStr(c.conn), c.cid)
dbgString := fmt.Sprintf("%s connection closed", c.typeString())
Debug(dbgString, clientConnStr(c.nc), c.cid)
c.clearAuthTimer()
c.clearPingTimer()
c.clearConnection()
// Snapshot for use.
subs := c.subs.All()
srv := c.srv
c.mu.Unlock()
if srv != nil {
// Unregister
srv.removeClient(c)
// Remove subscriptions.
// Remove clients subscriptions.
for _, s := range subs {
if sub, ok := s.(*subscription); ok {
srv.sl.Remove(sub.subject, sub)
}
}
}
// Check for a solicited route. If it was, start up a reconnect.
if c.isSolicitedRoute() {
go srv.connectToRoute(c.route.url)
}
}

13
server/client_test.go
View File

@@ -88,7 +88,6 @@ func TestClientCreateAndInfo(t *testing.T) {
info.Port != DEFAULT_PORT {
t.Fatalf("INFO inconsistent: %+v\n", info)
}
}
func TestClientConnect(t *testing.T) {
@@ -285,7 +284,7 @@ func TestClientPubWithQueueSub(t *testing.T) {
go func() {
c.parse(op)
c.conn.Close()
c.nc.Close()
}()
var n1, n2, received int
@@ -332,7 +331,7 @@ func TestClientUnSub(t *testing.T) {
go func() {
c.parse(op)
c.conn.Close()
c.nc.Close()
}()
var received int
@@ -373,7 +372,7 @@ func TestClientUnSubMax(t *testing.T) {
go func() {
c.parse(op)
c.conn.Close()
c.nc.Close()
}()
var received int
@@ -396,7 +395,7 @@ func TestClientUnSubMax(t *testing.T) {
func TestClientAutoUnsubExactReceived(t *testing.T) {
_, c, _ := setupClient()
defer c.conn.Close()
defer c.nc.Close()
// SUB/PUB
subs := []byte("SUB foo 1\r\n")
@@ -426,7 +425,7 @@ func TestClientAutoUnsubExactReceived(t *testing.T) {
func TestClientUnsubAfterAutoUnsub(t *testing.T) {
_, c, _ := setupClient()
defer c.conn.Close()
defer c.nc.Close()
// SUB/UNSUB/UNSUB
subs := []byte("SUB foo 1\r\n")
@@ -476,7 +475,7 @@ func TestClientRemoveSubsOnDisconnect(t *testing.T) {
func TestClientMapRemoval(t *testing.T) {
s, c, _ := setupClient()
c.conn.Close()
c.nc.Close()
end := time.Now().Add(1 * time.Second)
for time.Now().Before(end) {

35
server/configs/cluster.conf Normal file
View File

@@ -0,0 +1,35 @@
# Cluster config file
port: 4242
net: apcera.me # net interface
authorization {
user: derek
password: bella
timeout: 1
}
pid_file: '/tmp/nats_cluster_test.pid'
log_file: '/tmp/nats_cluster_test.log'
cluster {
host: '127.0.0.1'
port: 4244
authorization {
user: route_user
password: top_secret
timeout: 1
}
# Routes are actively solicited and connected to from this server.
# Other servers can connect to us if they supply the correct credentials
# in their routes definitions from above.
routes = [
nats-route://foo:bar@apcera.me:4245
nats-route://foo:bar@apcera.me:4246
]
}

2
server/connz.go
View File

@@ -39,7 +39,7 @@ func (s *Server) HandleConnz(w http.ResponseWriter, r *http.Request) {
InBytes: client.inBytes,
OutBytes: client.outBytes,
}
if ip, ok := client.conn.(*net.TCPConn); ok {
if ip, ok := client.nc.(*net.TCPConn); ok {
addr := ip.RemoteAddr().(*net.TCPAddr)
ci.Port = addr.Port
ci.Ip = addr.IP.String()

17
server/const.go
View File

@@ -7,7 +7,7 @@ import (
)
const (
VERSION = "go-0.2.12.alpha.1"
VERSION = "go-0.3.0.alpha.2"
DEFAULT_PORT = 4222
DEFAULT_HOST = "0.0.0.0"
@@ -44,4 +44,19 @@ const (
// Accept sleep times on temporary errors
ACCEPT_MIN_SLEEP = 10 * time.Millisecond
ACCEPT_MAX_SLEEP = 1 * time.Second
// Route solicitation intervals.
DEFAULT_ROUTE_CONNECT = 1 * time.Second
// Route dial timeout
DEFAULT_ROUTE_DIAL = 1 * time.Second
// Default size of proto to print on parse errors
PROTO_SNIPPET_SIZE = 32
// Maximum number of arguments from MSG proto
MAX_MSG_ARGS = 4
// Maximum number of arguments from PUB proto
MAX_PUB_ARGS = 3
)

129
server/opts.go
View File

@@ -3,7 +3,10 @@
package server
import (
"fmt"
"io/ioutil"
"net/url"
"strings"
"time"
@@ -11,24 +14,36 @@ import (
)
type Options struct {
Host string `json:"addr"`
Port int `json:"port"`
Trace bool `json:"-"`
Debug bool `json:"-"`
NoLog bool `json:"-"`
NoSigs bool `json:"-"`
Logtime bool `json:"-"`
MaxConn int `json:"max_connections"`
Username string `json:"user,omitempty"`
Password string `json:"-"`
Authorization string `json:"-"`
PingInterval time.Duration `json:"ping_interval"`
MaxPingsOut int `json:"ping_max"`
HttpPort int `json:"http_port"`
SslTimeout float64 `json:"ssl_timeout"`
AuthTimeout float64 `json:"auth_timeout"`
MaxControlLine int `json:"max_control_line"`
MaxPayload int `json:"max_payload"`
Host string `json:"addr"`
Port int `json:"port"`
Trace bool `json:"-"`
Debug bool `json:"-"`
NoLog bool `json:"-"`
NoSigs bool `json:"-"`
Logtime bool `json:"-"`
MaxConn int `json:"max_connections"`
Username string `json:"user,omitempty"`
Password string `json:"-"`
Authorization string `json:"-"`
PingInterval time.Duration `json:"ping_interval"`
MaxPingsOut int `json:"ping_max"`
HttpPort int `json:"http_port"`
SslTimeout float64 `json:"ssl_timeout"`
AuthTimeout float64 `json:"auth_timeout"`
MaxControlLine int `json:"max_control_line"`
MaxPayload int `json:"max_payload"`
ClusterHost string `json:"addr"`
ClusterPort int `json:"port"`
ClusterUsername string `json:"-"`
ClusterPassword string `json:"-"`
ClusterAuthTimeout float64 `json:"auth_timeout"`
Routes []*url.URL `json:"-"`
}
type authorization struct {
user string
pass string
timeout float64
}
// FIXME(dlc): Hacky
@@ -63,28 +78,73 @@ func ProcessConfigFile(configFile string) (*Options, error) {
opts.Logtime = v.(bool)
case "authorization":
am := v.(map[string]interface{})
for mk, mv := range am {
switch strings.ToLower(mk) {
case "user", "username":
opts.Username = mv.(string)
case "pass", "password":
opts.Password = mv.(string)
case "timeout":
at := float64(1)
switch mv.(type) {
case int64:
at = float64(mv.(int64))
case float64:
at = mv.(float64)
}
opts.AuthTimeout = at / float64(time.Second)
}
auth := parseAuthorization(am)
opts.Username = auth.user
opts.Password = auth.pass
opts.AuthTimeout = auth.timeout
case "cluster":
cm := v.(map[string]interface{})
if err := parseCluster(cm, opts); err != nil {
return nil, err
}
}
}
return opts, nil
}
// parseCluster will parse the cluster config.
func parseCluster(cm map[string]interface{}, opts *Options) error {
for mk, mv := range cm {
switch strings.ToLower(mk) {
case "port":
opts.ClusterPort = int(mv.(int64))
case "host", "net":
opts.ClusterHost = mv.(string)
case "authorization":
am := mv.(map[string]interface{})
auth := parseAuthorization(am)
opts.ClusterUsername = auth.user
opts.ClusterPassword = auth.pass
opts.ClusterAuthTimeout = auth.timeout
case "routes":
ra := mv.([]interface{})
opts.Routes = make([]*url.URL, 0, len(ra))
for _, r := range ra {
routeUrl := r.(string)
url, err := url.Parse(routeUrl)
if err != nil {
return fmt.Errorf("Error parsing route url [%q]", routeUrl)
}
opts.Routes = append(opts.Routes, url)
}
}
}
return nil
}
// Helper function to parse Authorization configs.
func parseAuthorization(am map[string]interface{}) authorization {
auth := authorization{}
for mk, mv := range am {
switch strings.ToLower(mk) {
case "user", "username":
auth.user = mv.(string)
case "pass", "password":
auth.pass = mv.(string)
case "timeout":
at := float64(1)
switch mv.(type) {
case int64:
at = float64(mv.(int64))
case float64:
at = mv.(float64)
}
auth.timeout = at
}
}
return auth
}
// Will merge two options giving preference to the flagOpts if the item is present.
func MergeOptions(fileOpts, flagOpts *Options) *Options {
if fileOpts == nil {
@@ -95,6 +155,7 @@ func MergeOptions(fileOpts, flagOpts *Options) *Options {
}
// Merge the two, flagOpts override
opts := *fileOpts
if flagOpts.Port != 0 {
opts.Port = flagOpts.Port
}

4
server/opts_test.go
View File

@@ -35,7 +35,7 @@ func TestConfigFile(t *testing.T) {
Port: 4242,
Username: "derek",
Password: "bella",
AuthTimeout: 1.0 / float64(time.Second),
AuthTimeout: 1.0,
Debug: false,
Trace: true,
Logtime: false,
@@ -57,7 +57,7 @@ func TestMergeOverrides(t *testing.T) {
Port: 2222,
Username: "derek",
Password: "spooky",
AuthTimeout: 1.0 / float64(time.Second),
AuthTimeout: 1.0,
Debug: true,
Trace: true,
Logtime: false,

75
server/parser.go
View File

@@ -9,6 +9,7 @@ import (
type pubArg struct {
subject []byte
reply []byte
sid []byte
szb []byte
size int
}
@@ -57,6 +58,11 @@ const (
OP_UNSU
OP_UNSUB
UNSUB_ARG
OP_M
OP_MS
OP_MSG
OP_MSG_SPC
MSG_ARG
)
func (c *client) parse(buf []byte) error {
@@ -78,6 +84,8 @@ func (c *client) parse(buf []byte) error {
c.state = OP_S
case 'U', 'u':
c.state = OP_U
case 'M', 'm':
c.state = OP_M
default:
goto parseErr
}
@@ -106,7 +114,7 @@ func (c *client) parse(buf []byte) error {
default:
goto parseErr
}
case OP_PUB_SPC:
case OP_PUB_SPC:
switch b {
case ' ', '\t':
continue
@@ -360,6 +368,55 @@ func (c *client) parse(buf []byte) error {
}
c.drop, c.state = 0, OP_START
}
case OP_M:
switch b {
case 'S', 's':
c.state = OP_MS
default:
goto parseErr
}
case OP_MS:
switch b {
case 'G', 'g':
c.state = OP_MSG
default:
goto parseErr
}
case OP_MSG:
switch b {
case ' ', '\t':
c.state = OP_MSG_SPC
default:
goto parseErr
}
case OP_MSG_SPC:
switch b {
case ' ', '\t':
continue
default:
c.state = MSG_ARG
c.as = i
}
case MSG_ARG:
switch b {
case '\r':
c.drop = 1
case '\n':
var arg []byte
if c.argBuf != nil {
arg = c.argBuf
} else {
arg = buf[c.as : i-c.drop]
}
if err := c.processMsgArgs(arg); err != nil {
return err
}
c.drop, c.as, c.state = 0, i+1, MSG_PAYLOAD
default:
if c.argBuf != nil {
c.argBuf = append(c.argBuf, b)
}
}
default:
goto parseErr
}
@@ -390,13 +447,19 @@ authErr:
parseErr:
c.sendErr("Unknown Protocol Operation")
stop := i + 32
if stop > len(buf) {
stop = len(buf)-1
}
return fmt.Errorf("Parse Error, state=%d,i=%d: '%s'", c.state, i, buf[i:stop])
snip := protoSnippet(i, buf)
err := fmt.Errorf("%s Parser ERROR, state=%d, i=%d: proto='%s...'",
c.typeString(), c.state, i, snip)
return err
}
func protoSnippet(start int, buf []byte) string {
stop := start + PROTO_SNIPPET_SIZE
if stop > len(buf) {
stop = len(buf) - 1
}
return fmt.Sprintf("%q", buf[start:stop])
}
// clonePubArg is used when the split buffer scenario has the pubArg in the existing read buffer, but
// we need to hold onto it into the next read.

80
server/parser_test.go
View File

@@ -1,4 +1,5 @@
// Copyright 2012 Apcera Inc. All rights reserved.
// Copyright 2012-2013 Apcera Inc. All rights reserved.
package server
@@ -222,6 +223,83 @@ func TestParsePubArg(t *testing.T) {
testPubArg(c, t)
}
func TestParseMsg(t *testing.T) {
c := dummyClient()
pub := []byte("MSG foo RSID:1:2 5\r\nhello\r")
err := c.parse(pub)
if err != nil || c.state != MSG_END {
t.Fatalf("Unexpected: %d : %v\n", c.state, err)
}
if !bytes.Equal(c.pa.subject, []byte("foo")) {
t.Fatalf("Did not parse subject correctly: 'foo' vs '%s'\n", c.pa.subject)
}
if c.pa.reply != nil {
t.Fatalf("Did not parse reply correctly: 'nil' vs '%s'\n", c.pa.reply)
}
if c.pa.size != 5 {
t.Fatalf("Did not parse msg size correctly: 5 vs %d\n", c.pa.size)
}
if !bytes.Equal(c.pa.sid, []byte("RSID:1:2")) {
t.Fatalf("Did not parse sid correctly: 'RSID:1:2' vs '%s'\n", c.pa.sid)
}
c.state = OP_START
pub = []byte("MSG foo.bar RSID:1:2 INBOX.22 11\r\nhello world\r")
err = c.parse(pub)
if err != nil || c.state != MSG_END {
t.Fatalf("Unexpected: %d : %v\n", c.state, err)
}
if !bytes.Equal(c.pa.subject, []byte("foo.bar")) {
t.Fatalf("Did not parse subject correctly: 'foo' vs '%s'\n", c.pa.subject)
}
if !bytes.Equal(c.pa.reply, []byte("INBOX.22")) {
t.Fatalf("Did not parse reply correctly: 'INBOX.22' vs '%s'\n", c.pa.reply)
}
if c.pa.size != 11 {
t.Fatalf("Did not parse msg size correctly: 11 vs %d\n", c.pa.size)
}
}
func testMsgArg(c *client, t *testing.T) {
if !bytes.Equal(c.pa.subject, []byte("foobar")) {
t.Fatalf("Mismatched subject: '%s'\n", c.pa.subject)
}
if !bytes.Equal(c.pa.szb, []byte("22")) {
t.Fatalf("Bad size buf: '%s'\n", c.pa.szb)
}
if c.pa.size != 22 {
t.Fatalf("Bad size: %d\n", c.pa.size)
}
if !bytes.Equal(c.pa.sid, []byte("RSID:22:1")) {
t.Fatalf("Bad sid: '%s'\n", c.pa.sid)
}
}
func TestParseMsgArg(t *testing.T) {
c := dummyClient()
if err := c.processMsgArgs([]byte("foobar RSID:22:1 22")) ; err != nil {
t.Fatalf("Unexpected parse error: %v\n", err)
}
testMsgArg(c, t)
if err := c.processMsgArgs([]byte(" foobar RSID:22:1 22")) ; err != nil {
t.Fatalf("Unexpected parse error: %v\n", err)
}
testMsgArg(c, t)
if err := c.processMsgArgs([]byte(" foobar RSID:22:1 22 ")) ; err != nil {
t.Fatalf("Unexpected parse error: %v\n", err)
}
testMsgArg(c, t)
if err := c.processMsgArgs([]byte("foobar RSID:22:1 \t22")) ; err != nil {
t.Fatalf("Unexpected parse error: %v\n", err)
}
if err := c.processMsgArgs([]byte("foobar\t\tRSID:22:1\t22\r")) ; err != nil {
t.Fatalf("Unexpected parse error: %v\n", err)
}
testMsgArg(c, t)
}
func TestShouldFail(t *testing.T) {
c := dummyClient()

296
server/route.go Normal file
View File

@@ -0,0 +1,296 @@
// Copyright 2013 Apcera Inc. All rights reserved.
package server
import (
"bytes"
"encoding/json"
"fmt"
"net"
"net/url"
"regexp"
"time"
)
type route struct {
remoteId string
didSolicit bool
url *url.URL
}
type connectInfo struct {
Verbose bool `json:"verbose"`
Pedantic bool `json:"pedantic"`
User string `json:"user,omitempty"`
Pass string `json:"pass,omitempty"`
Ssl bool `json:"ssl_required"`
Name string `json:"name"`
}
const conProto = "CONNECT %s" + _CRLF_
func (c *client) sendConnect() {
var user, pass string
if userInfo := c.route.url.User; userInfo != nil {
user = userInfo.Username()
pass, _ = userInfo.Password()
}
cinfo := connectInfo{
Verbose: false,
Pedantic: false,
User: user,
Pass: pass,
Ssl: false,
Name: c.srv.info.Id,
}
b, err := json.Marshal(cinfo)
if err != nil {
Logf("Error marshalling CONNECT to route: %v\n", err)
c.closeConnection()
}
c.bw.WriteString(fmt.Sprintf(conProto, b))
c.bw.Flush()
}
func (s *Server) sendLocalSubsToRoute(route *client) {
for _, client := range s.clients {
for _, s := range client.subs.All() {
if sub, ok := s.(*subscription); ok {
rsid := routeSid(sub)
proto := fmt.Sprintf(subProto, sub.subject, sub.queue, rsid)
route.bw.WriteString(proto)
}
}
}
route.bw.Flush()
Debug("Route sent local subscriptions", clientConnStr(route.nc), route.cid)
}
func (s *Server) createRoute(conn net.Conn, rUrl *url.URL) *client {
didSolicit := rUrl != nil
r := &route{didSolicit: didSolicit}
c := &client{srv: s, nc: conn, opts: defaultOpts, typ: ROUTER, route: r}
// Initialize
c.initClient()
Debug("Route connection created", clientConnStr(c.nc), c.cid)
c.mu.Lock()
// Queue Connect proto if we solicited the connection.
if didSolicit {
r.url = rUrl
Debug("Route connect msg sent", clientConnStr(c.nc), c.cid)
c.sendConnect()
}
// Send our info to the other side.
s.sendInfo(c)
// Check for Auth required state for incoming connections.
if s.routeInfo.AuthRequired && !didSolicit {
ttl := secondsToDuration(s.opts.ClusterAuthTimeout)
c.setAuthTimer(ttl)
}
c.mu.Unlock()
// Register with the server.
s.mu.Lock()
s.routes[c.cid] = c
s.mu.Unlock()
// Send our local subscriptions to this route.
s.sendLocalSubsToRoute(c)
return c
}
const (
_CRLF_ = "\r\n"
_EMPTY_ = ""
_SPC_ = " "
)
const (
subProto = "SUB %s %s %s" + _CRLF_
unsubProto = "UNSUB %s%s" + _CRLF_
)
// FIXME(dlc) - Make these reserved and reject if they come in as a sid
// from a client connection.
const (
RSID = "RSID"
QRSID = "QRSID"
RSID_CID_INDEX = 1
RSID_SID_INDEX = 2
EXPECTED_MATCHES = 3
)
// FIXME(dlc) - This may be too slow, check at later date.
var qrsidRe = regexp.MustCompile(`QRSID:(\d+):([^\s]+)`)
func (s *Server) routeSidQueueSubscriber(rsid []byte) *subscription {
if !bytes.HasPrefix(rsid, []byte(QRSID)) {
return nil
}
matches := qrsidRe.FindSubmatch(rsid)
if matches == nil || len(matches) != EXPECTED_MATCHES {
return nil
}
cid := uint64(parseInt64(matches[RSID_CID_INDEX]))
client := s.clients[cid]
if client == nil {
return nil
}
sid := matches[RSID_SID_INDEX]
if sub, ok := (client.subs.Get(sid)).(*subscription); ok {
return sub
}
return nil
}
func routeSid(sub *subscription) string {
var qi string
if len(sub.queue) > 0 {
qi = "Q"
}
return fmt.Sprintf("%s%s:%d:%s", qi, RSID, sub.client.cid, sub.sid)
}
func (s *Server) broadcastToRoutes(proto string) {
for _, route := range s.routes {
// FIXME(dlc) - Make same logic as deliverMsg
route.bw.WriteString(proto)
route.bw.Flush()
}
}
// broadcastSubscribe will forward a client subscription
// to all active routes.
func (s *Server) broadcastSubscribe(sub *subscription) {
rsid := routeSid(sub)
proto := fmt.Sprintf(subProto, sub.subject, sub.queue, rsid)
s.broadcastToRoutes(proto)
}
// broadcastUnSubscribe will forward a client unsubscribe
// action to all active routes.
func (s *Server) broadcastUnSubscribe(sub *subscription) {
rsid := routeSid(sub)
maxStr := _EMPTY_
if sub.max > 0 {
maxStr = fmt.Sprintf("%d ", sub.max)
}
proto := fmt.Sprintf(unsubProto, maxStr, rsid)
s.broadcastToRoutes(proto)
}
func (s *Server) routeAcceptLoop(ch chan struct{}) {
hp := fmt.Sprintf("%s:%d", s.opts.ClusterHost, s.opts.ClusterPort)
Logf("Listening for route connections on %s", hp)
l, e := net.Listen("tcp", hp)
if e != nil {
Fatalf("Error listening on router port: %d - %v", s.opts.Port, e)
return
}
// Let them know we are up
close(ch)
// Setup state that can enable shutdown
s.mu.Lock()
s.routeListener = l
s.mu.Unlock()
tmpDelay := ACCEPT_MIN_SLEEP
for s.isRunning() {
conn, err := l.Accept()
if err != nil {
if ne, ok := err.(net.Error); ok && ne.Temporary() {
Debug("Temporary Route Accept Error(%v), sleeping %dms",
ne, tmpDelay/time.Millisecond)
time.Sleep(tmpDelay)
tmpDelay *= 2
if tmpDelay > ACCEPT_MAX_SLEEP {
tmpDelay = ACCEPT_MAX_SLEEP
}
} else if s.isRunning() {
Logf("Accept error: %v", err)
}
continue
}
tmpDelay = ACCEPT_MIN_SLEEP
s.createRoute(conn, nil)
}
Debug("Router accept loop exiting..")
s.done <- true
}
// StartRouting will start the accept loop on the cluster host:port
// and will actively try to connect to listed routes.
func (s *Server) StartRouting() {
info := Info{
Id: s.info.Id,
Version: s.info.Version,
Host: s.opts.ClusterHost,
Port: s.opts.ClusterPort,
AuthRequired: false,
SslRequired: false,
MaxPayload: MAX_PAYLOAD_SIZE,
}
// Check for Auth items
if s.opts.ClusterUsername != "" {
info.AuthRequired = true
}
s.routeInfo = info
// Generate the info json
b, err := json.Marshal(info)
if err != nil {
Fatalf("Error marshalling Route INFO JSON: %+v\n", err)
}
s.routeInfoJson = []byte(fmt.Sprintf("INFO %s %s", b, CR_LF))
// Spin up the accept loop
ch := make(chan struct{})
go s.routeAcceptLoop(ch)
<-ch
// Solicit Routes if needed.
s.solicitRoutes()
}
func (s *Server) connectToRoute(rUrl *url.URL) {
for s.isRunning() {
Debugf("Trying to connect to route on %s", rUrl.Host)
conn, err := net.DialTimeout("tcp", rUrl.Host, DEFAULT_ROUTE_DIAL)
if err != nil {
Debugf("Error trying to connect to route: %v", err)
select {
case <-s.rcQuit:
return
case <-time.After(DEFAULT_ROUTE_CONNECT):
continue
}
}
// We have a route connection here.
// Go ahead and create it and exit this func.
s.createRoute(conn, rUrl)
return
}
}
func (c *client) isSolicitedRoute() bool {
c.mu.Lock()
defer c.mu.Unlock()
return c.typ == ROUTER && c.route != nil && c.route.didSolicit
}
func (s *Server) solicitRoutes() {
for _, r := range s.opts.Routes {
go s.connectToRoute(r)
}
}

40
server/routes_test.go Normal file
View File

@@ -0,0 +1,40 @@
// Copyright 2013 Apcera Inc. All rights reserved.
package server
import (
"net/url"
"reflect"
"testing"
)
func TestRouteConfig(t *testing.T) {
opts, err := ProcessConfigFile("./configs/cluster.conf")
if err != nil {
t.Fatalf("Received an error reading route config file: %v\n", err)
}
golden := &Options{
Host: "apcera.me",
Port: 4242,
Username: "derek",
Password: "bella",
AuthTimeout: 1.0,
ClusterHost: "127.0.0.1",
ClusterPort: 4244,
ClusterUsername: "route_user",
ClusterPassword: "top_secret",
ClusterAuthTimeout: 1.0,
}
// Setup URLs
r1, _ := url.Parse("nats-route://foo:bar@apcera.me:4245")
r2, _ := url.Parse("nats-route://foo:bar@apcera.me:4246")
golden.Routes = []*url.URL{r1, r2}
if !reflect.DeepEqual(golden, opts) {
t.Fatalf("Options are incorrect from config file.\nexpected: %+v\ngot: %+v",
golden, opts)
}
}

133
server/server.go
View File

@@ -3,7 +3,6 @@
package server
import (
"bufio"
"encoding/json"
"fmt"
"net"
@@ -11,10 +10,8 @@ import (
"os"
"os/signal"
"sync"
"sync/atomic"
"time"
"github.com/apcera/gnatsd/hashmap"
"github.com/apcera/gnatsd/sublist"
)
@@ -40,9 +37,16 @@ type Server struct {
running bool
listener net.Listener
clients map[uint64]*client
routes map[uint64]*client
done chan bool
start time.Time
stats
routeListener net.Listener
grid uint64
routeInfo Info
routeInfoJson []byte
rcQuit chan bool
}
type stats struct {
@@ -83,18 +87,29 @@ func New(opts *Options) *Server {
// Setup logging with flags
s.LogInit()
// For tracing clients
// For tracking clients
s.clients = make(map[uint64]*client)
// For tracking routes
s.routes = make(map[uint64]*client)
// Used to kick out all of the route
// connect Go routines.
s.rcQuit = make(chan bool)
// Generate the info json
b, err := json.Marshal(s.info)
if err != nil {
Fatalf("Err marshalling INFO JSON: %+v\n", err)
Fatalf("Error marshalling INFO JSON: %+v\n", err)
}
s.infoJson = []byte(fmt.Sprintf("INFO %s %s", b, CR_LF))
s.handleSignals()
Logf("Starting nats-server version %s", VERSION)
s.running = true
return s
}
@@ -140,11 +155,26 @@ func (s *Server) Shutdown() {
clients[i] = c
}
// Kick AcceptLoop()
// Number of done channel responses we expect.
doneExpected := 0
// Kick client AcceptLoop()
if s.listener != nil {
doneExpected++
s.listener.Close()
s.listener = nil
}
// Kick route AcceptLoop()
if s.routeListener != nil {
doneExpected++
s.routeListener.Close()
s.routeListener = nil
}
// Release the solicited routes connect go routines.
close(s.rcQuit)
s.mu.Unlock()
// Close client connections
@@ -152,13 +182,16 @@ func (s *Server) Shutdown() {
c.closeConnection()
}
<-s.done
// Block until the accept loops exit
for doneExpected > 0 {
<-s.done
doneExpected--
}
}
func (s *Server) AcceptLoop() {
Logf("Starting nats-server version %s on port %d", VERSION, s.opts.Port)
hp := fmt.Sprintf("%s:%d", s.opts.Host, s.opts.Port)
Logf("Listening for client connections on %s", hp)
l, e := net.Listen("tcp", hp)
if e != nil {
Fatalf("Error listening on port: %d - %v", s.opts.Port, e)
@@ -170,7 +203,6 @@ func (s *Server) AcceptLoop() {
// Setup state that can enable shutdown
s.mu.Lock()
s.listener = l
s.running = true
s.mu.Unlock()
tmpDelay := ACCEPT_MIN_SLEEP
@@ -179,14 +211,14 @@ func (s *Server) AcceptLoop() {
conn, err := l.Accept()
if err != nil {
if ne, ok := err.(net.Error); ok && ne.Temporary() {
Debug("Temporary Accept Error(%v), sleeping %dms",
Debug("Temporary Client Accept Error(%v), sleeping %dms",
ne, tmpDelay/time.Millisecond)
time.Sleep(tmpDelay)
tmpDelay *= 2
if tmpDelay > ACCEPT_MAX_SLEEP {
tmpDelay = ACCEPT_MAX_SLEEP
}
} else {
} else if s.isRunning() {
Logf("Accept error: %v", err)
}
continue
@@ -194,8 +226,8 @@ func (s *Server) AcceptLoop() {
tmpDelay = ACCEPT_MIN_SLEEP
s.createClient(conn)
}
s.done <- true
Log("Server Exiting..")
s.done <- true
}
func (s *Server) StartHTTPMonitoring() {
@@ -219,33 +251,24 @@ func (s *Server) StartHTTPMonitoring() {
}
func (s *Server) createClient(conn net.Conn) *client {
c := &client{srv: s, conn: conn, opts: defaultOpts}
c := &client{srv: s, nc: conn, opts: defaultOpts}
// Initialize
c.initClient()
Debug("Client connection created", clientConnStr(c.nc), c.cid)
c.mu.Lock()
c.cid = atomic.AddUint64(&s.gcid, 1)
c.bw = bufio.NewWriterSize(c.conn, defaultBufSize)
c.subs = hashmap.New()
// This is to track pending clients that have data to be flushed
// after we process inbound msgs from our own connection.
c.pcd = make(map[*client]struct{})
Debug("Client connection created", clientConnStr(conn), c.cid)
if ip, ok := conn.(*net.TCPConn); ok {
ip.SetReadBuffer(defaultBufSize)
}
// Send our information.
s.sendInfo(c)
go c.readLoop()
// Check for Auth
if s.info.AuthRequired {
c.setAuthTimer(AUTH_TIMEOUT) // FIXME(dlc): Make option
ttl := secondsToDuration(s.opts.AuthTimeout)
c.setAuthTimer(ttl)
}
// Set the Ping timer
c.setPingTimer()
c.mu.Unlock()
// Register with the server.
@@ -257,12 +280,15 @@ func (s *Server) createClient(conn net.Conn) *client {
}
func (s *Server) sendInfo(c *client) {
// FIXME, err
c.conn.Write(s.infoJson)
switch c.typ {
case CLIENT:
c.nc.Write(s.infoJson)
case ROUTER:
c.nc.Write(s.routeInfoJson)
}
}
// Check auth and return boolean indicating if client is ok
func (s *Server) checkAuth(c *client) bool {
func (s *Server) checkClientAuth(c *client) bool {
if !s.info.AuthRequired {
return true
}
@@ -277,8 +303,41 @@ func (s *Server) checkAuth(c *client) bool {
return true
}
func (s *Server) checkRouterAuth(c *client) bool {
if !s.routeInfo.AuthRequired {
return true
}
if s.opts.ClusterUsername != c.opts.Username ||
s.opts.ClusterPassword != c.opts.Password {
return false
}
return true
}
// Check auth and return boolean indicating if client is ok
func (s *Server) checkAuth(c *client) bool {
switch c.typ {
case CLIENT:
return s.checkClientAuth(c)
case ROUTER:
return s.checkRouterAuth(c)
default:
return false
}
}
func (s *Server) removeClient(c *client) {
c.mu.Lock()
cid := c.cid
typ := c.typ
c.mu.Unlock()
s.mu.Lock()
delete(s.clients, c.cid)
switch typ {
case CLIENT:
delete(s.clients, cid)
case ROUTER:
delete(s.routes, cid)
}
s.mu.Unlock()
}

21
server/util.go
View File

@@ -6,6 +6,7 @@ import (
"crypto/rand"
"encoding/hex"
"io"
"time"
)
func genId() string {
@@ -34,3 +35,23 @@ func parseSize(d []byte) (n int) {
}
return n
}
// parseInt64 expects decimal positive numbers. We
// return -1 to signal error
func parseInt64(d []byte) (n int64) {
if len(d) == 0 {
return -1
}
for _, dec := range d {
if dec < ascii_0 || dec > ascii_9 {
return -1
}
n = n*10 + (int64(dec) - ascii_0)
}
return n
}
func secondsToDuration(seconds float64) time.Duration {
ttl := seconds * float64(time.Second)
return time.Duration(ttl)
}

26
test/configs/cluster.conf Normal file
View File

@@ -0,0 +1,26 @@
# Cluster config file
port: 4242
#net: apcera.me # net interface
cluster {
host: '127.0.0.1'
port: 4244
authorization {
user: route_user
password: top_secret
timeout: 0.5
}
# Routes are actively solicited and connected to from this server.
# Other servers can connect to us if they supply the correct credentials
# in their routes definitions from above.
routes = [
nats-route://foo:bar@127.0.0.1:4245
nats-route://foo:bar@127.0.0.1:4246
]
}

11
test/gosrv_test.go
View File

@@ -9,26 +9,27 @@ import (
)
func TestSimpleGoServerShutdown(t *testing.T) {
s := runDefaultServer()
base := runtime.NumGoroutine()
s := runDefaultServer()
s.Shutdown()
time.Sleep(10 * time.Millisecond)
delta := (runtime.NumGoroutine() - base)
if delta > 0 {
if delta > 1 {
t.Fatalf("%d Go routines still exist post Shutdown()", delta)
}
}
func TestGoServerShutdownWithClients(t *testing.T) {
base := runtime.NumGoroutine()
s := runDefaultServer()
for i := 0 ; i < 10 ; i++ {
createClientConn(t, "localhost", 4222)
}
base := runtime.NumGoroutine()
s.Shutdown()
time.Sleep(10 * time.Millisecond)
// Wait longer for client connections
time.Sleep(50 * time.Millisecond)
delta := (runtime.NumGoroutine() - base)
if delta > 0 {
if delta > 1 {
t.Fatalf("%d Go routines still exist post Shutdown()", delta)
}
}

2
test/proto_test.go
View File

@@ -98,7 +98,7 @@ func TestQueueSub(t *testing.T) {
}
for k, c := range sids {
if c < 35 {
t.Fatalf("Expected ~50 (+-15) msgs for '%s', got %d\n", k, c)
t.Fatalf("Expected ~50 (+-15) msgs for sid:'%s', got %d\n", k, c)
}
}
}

427
test/routes_test.go Normal file
View File

@@ -0,0 +1,427 @@
// Copyright 2012 Apcera Inc. All rights reserved.
package test
import (
"encoding/json"
"fmt"
"io/ioutil"
"runtime"
"strings"
"testing"
"time"
"github.com/apcera/gnatsd/server"
)
func runRouteServer(t *testing.T) (*server.Server, *server.Options) {
opts, err := server.ProcessConfigFile("./configs/cluster.conf")
// Override for running in Go routine.
opts.NoSigs = true
opts.Debug = true
opts.Trace = true
opts.NoLog = true
if err != nil {
t.Fatalf("Error parsing config file: %v\n", err)
}
return RunServer(opts), opts
}
func TestRouterListeningSocket(t *testing.T) {
s, opts := runRouteServer(t)
defer s.Shutdown()
// Check that the cluster socket is able to be connected.
addr := fmt.Sprintf("%s:%d", opts.ClusterHost, opts.ClusterPort)
checkSocket(t, addr, 2*time.Second)
}
func TestRouteGoServerShutdown(t *testing.T) {
base := runtime.NumGoroutine()
s, _ := runRouteServer(t)
s.Shutdown()
time.Sleep(10 * time.Millisecond)
delta := (runtime.NumGoroutine() - base)
if delta > 1 {
panic("foo")
t.Fatalf("%d Go routines still exist post Shutdown()", delta)
}
}
func TestSendRouteInfoOnConnect(t *testing.T) {
s, opts := runRouteServer(t)
defer s.Shutdown()
rc := createRouteConn(t, opts.ClusterHost, opts.ClusterPort)
_, expect := setupRoute(t, rc, opts)
buf := expect(infoRe)
info := server.Info{}
if err := json.Unmarshal(buf[4:], &info); err != nil {
t.Fatalf("Could not unmarshal route info: %v", err)
}
if !info.AuthRequired {
t.Fatal("Expected to see AuthRequired")
}
if info.Port != opts.ClusterPort {
t.Fatalf("Received wrong information for port, expected %d, got %d",
info.Port, opts.ClusterPort)
}
}
func TestSendRouteSubAndUnsub(t *testing.T) {
s, opts := runRouteServer(t)
defer s.Shutdown()
c := createClientConn(t, opts.Host, opts.Port)
defer c.Close()
send, _ := setupConn(t, c)
// We connect to the route.
rc := createRouteConn(t, opts.ClusterHost, opts.ClusterPort)
expectAuthRequired(t, rc)
setupRoute(t, rc, opts)
// Send SUB via client connection
send("SUB foo 22\r\n")
// Make sure the SUB is broadcast via the route
buf := expectResult(t, rc, subRe)
matches := subRe.FindAllSubmatch(buf, -1)
rsid := string(matches[0][5])
if !strings.HasPrefix(rsid, "RSID:") {
t.Fatalf("Got wrong RSID: %s\n", rsid)
}
// Send UNSUB via client connection
send("UNSUB 22\r\n")
// Make sure the SUB is broadcast via the route
buf = expectResult(t, rc, unsubRe)
matches = unsubRe.FindAllSubmatch(buf, -1)
rsid2 := string(matches[0][1])
if rsid2 != rsid {
t.Fatalf("Expected rsid's to match. %q vs %q\n", rsid, rsid2)
}
}
func TestSendRouteSolicit(t *testing.T) {
s, opts := runRouteServer(t)
defer s.Shutdown()
// Listen for a connection from the server on the first route.
if len(opts.Routes) <= 0 {
t.Fatalf("Need an outbound solicted route for this test")
}
rUrl := opts.Routes[0]
conn := acceptRouteConn(t, rUrl.Host, server.DEFAULT_ROUTE_CONNECT)
defer conn.Close()
// We should receive a connect message right away due to auth.
buf := expectResult(t, conn, connectRe)
// Check INFO follows. Could be inline, with first result, if not
// check follow-on buffer.
if !infoRe.Match(buf) {
expectResult(t, conn, infoRe)
}
}
func TestRouteForwardsMsgFromClients(t *testing.T) {
s, opts := runRouteServer(t)
defer s.Shutdown()
client := createClientConn(t, opts.Host, opts.Port)
defer client.Close()
clientSend, clientExpect := setupConn(t, client)
route := acceptRouteConn(t, opts.Routes[0].Host, server.DEFAULT_ROUTE_CONNECT)
defer route.Close()
routeSend, routeExpect := setupRoute(t, route, opts)
expectMsgs := expectMsgsCommand(t, routeExpect)
// Eat the CONNECT and INFO protos
routeExpect(infoRe)
// Send SUB via route connection
routeSend("SUB foo RSID:2:22\r\n")
routeSend("PING\r\n")
routeExpect(pongRe)
// Send PUB via client connection
clientSend("PUB foo 2\r\nok\r\n")
clientSend("PING\r\n")
clientExpect(pongRe)
matches := expectMsgs(1)
checkMsg(t, matches[0], "foo", "RSID:2:22", "", "2", "ok")
}
func TestRouteForwardsMsgToClients(t *testing.T) {
s, opts := runRouteServer(t)
defer s.Shutdown()
client := createClientConn(t, opts.Host, opts.Port)
defer client.Close()
clientSend, clientExpect := setupConn(t, client)
expectMsgs := expectMsgsCommand(t, clientExpect)
route := createRouteConn(t, opts.ClusterHost, opts.ClusterPort)
expectAuthRequired(t, route)
routeSend, _ := setupRoute(t, route, opts)
// Subscribe to foo
clientSend("SUB foo 1\r\n")
// Use ping roundtrip to make sure its processed.
clientSend("PING\r\n")
clientExpect(pongRe)
// Send MSG proto via route connection
routeSend("MSG foo 1 2\r\nok\r\n")
matches := expectMsgs(1)
checkMsg(t, matches[0], "foo", "1", "", "2", "ok")
}
func TestRouteOneHopSemantics(t *testing.T) {
s, opts := runRouteServer(t)
defer s.Shutdown()
route := createRouteConn(t, opts.ClusterHost, opts.ClusterPort)
expectAuthRequired(t, route)
routeSend, _ := setupRoute(t, route, opts)
// Express interest on this route for foo.
routeSend("SUB foo RSID:2:2\r\n")
// Send MSG proto via route connection
routeSend("MSG foo 1 2\r\nok\r\n")
// Make sure it does not come back!
expectNothing(t, route)
}
func TestRouteOnlySendOnce(t *testing.T) {
s, opts := runRouteServer(t)
defer s.Shutdown()
client := createClientConn(t, opts.Host, opts.Port)
defer client.Close()
clientSend, clientExpect := setupConn(t, client)
route := createRouteConn(t, opts.ClusterHost, opts.ClusterPort)
expectAuthRequired(t, route)
routeSend, routeExpect := setupRoute(t, route, opts)
expectMsgs := expectMsgsCommand(t, routeExpect)
// Express multiple interest on this route for foo.
routeSend("SUB foo RSID:2:1\r\n")
routeSend("SUB foo RSID:2:2\r\n")
routeSend("PING\r\n")
routeExpect(pongRe)
// Send PUB via client connection
clientSend("PUB foo 2\r\nok\r\n")
clientSend("PING\r\n")
clientExpect(pongRe)
matches := expectMsgs(1)
checkMsg(t, matches[0], "foo", "RSID:2:1", "", "2", "ok")
}
func TestRouteQueueSemantics(t *testing.T) {
s, opts := runRouteServer(t)
defer s.Shutdown()
client := createClientConn(t, opts.Host, opts.Port)
clientSend, clientExpect := setupConn(t, client)
clientExpectMsgs := expectMsgsCommand(t, clientExpect)
defer client.Close()
route := createRouteConn(t, opts.ClusterHost, opts.ClusterPort)
expectAuthRequired(t, route)
routeSend, routeExpect := setupRoute(t, route, opts)
expectMsgs := expectMsgsCommand(t, routeExpect)
// Express multiple interest on this route for foo, queue group bar.
qrsid1 := "RSID:2:1"
routeSend(fmt.Sprintf("SUB foo bar %s\r\n", qrsid1))
qrsid2 := "RSID:2:2"
routeSend(fmt.Sprintf("SUB foo bar %s\r\n", qrsid2))
// Use ping roundtrip to make sure its processed.
routeSend("PING\r\n")
routeExpect(pongRe)
// Send PUB via client connection
clientSend("PUB foo 2\r\nok\r\n")
// Use ping roundtrip to make sure its processed.
clientSend("PING\r\n")
clientExpect(pongRe)
// Only 1
matches := expectMsgs(1)
checkMsg(t, matches[0], "foo", "", "", "2", "ok")
// Add normal Interest as well to route interest.
routeSend("SUB foo RSID:2:4\r\n")
// Use ping roundtrip to make sure its processed.
routeSend("PING\r\n")
routeExpect(pongRe)
// Send PUB via client connection
clientSend("PUB foo 2\r\nok\r\n")
// Use ping roundtrip to make sure its processed.
clientSend("PING\r\n")
clientExpect(pongRe)
// Should be 2 now, 1 for all normal, and one for specific queue subscriber.
matches = expectMsgs(2)
// Expect first to be the normal subscriber, next will be the queue one.
checkMsg(t, matches[0], "foo", "RSID:2:4", "", "2", "ok")
checkMsg(t, matches[1], "foo", "", "", "2", "ok")
// Check the rsid to verify it is one of the queue group subscribers.
rsid := string(matches[1][SID_INDEX])
if rsid != qrsid1 && rsid != qrsid2 {
t.Fatalf("Expected a queue group rsid, got %s\n", rsid)
}
// Now create a queue subscription for the client as well as a normal one.
clientSend("SUB foo 1\r\n")
// Use ping roundtrip to make sure its processed.
clientSend("PING\r\n")
clientExpect(pongRe)
routeExpect(subRe)
clientSend("SUB foo bar 2\r\n")
// Use ping roundtrip to make sure its processed.
clientSend("PING\r\n")
clientExpect(pongRe)
routeExpect(subRe)
// Deliver a MSG from the route itself, make sure the client receives both.
routeSend("MSG foo RSID:2:1 2\r\nok\r\n")
// Queue group one.
routeSend("MSG foo QRSID:2:2 2\r\nok\r\n")
// Use ping roundtrip to make sure its processed.
routeSend("PING\r\n")
routeExpect(pongRe)
// Should be 2 now, 1 for all normal, and one for specific queue subscriber.
matches = clientExpectMsgs(2)
// Expect first to be the normal subscriber, next will be the queue one.
checkMsg(t, matches[0], "foo", "1", "", "2", "ok")
checkMsg(t, matches[1], "foo", "2", "", "2", "ok")
}
func TestSolicitRouteReconnect(t *testing.T) {
s, opts := runRouteServer(t)
defer s.Shutdown()
rUrl := opts.Routes[0]
route := acceptRouteConn(t, rUrl.Host, server.DEFAULT_ROUTE_CONNECT)
// Go ahead and close the Route.
route.Close()
// We expect to get called back..
route = acceptRouteConn(t, rUrl.Host, 2*server.DEFAULT_ROUTE_CONNECT)
}
func TestMultipleRoutesSameId(t *testing.T) {
s, opts := runRouteServer(t)
defer s.Shutdown()
route1 := createRouteConn(t, opts.ClusterHost, opts.ClusterPort)
expectAuthRequired(t, route1)
route1Send, _ := setupRouteEx(t, route1, opts, "ROUTE:2222")
route2 := createRouteConn(t, opts.ClusterHost, opts.ClusterPort)
expectAuthRequired(t, route2)
route2Send, _ := setupRouteEx(t, route2, opts, "ROUTE:2222")
// Send SUB via route connections
sub := "SUB foo RSID:2:22\r\n"
route1Send(sub)
route2Send(sub)
// Make sure we do not get anything on a MSG send to a router.
// Send MSG proto via route connection
route1Send("MSG foo 1 2\r\nok\r\n")
expectNothing(t, route1)
expectNothing(t, route2)
// Setup a client
client := createClientConn(t, opts.Host, opts.Port)
clientSend, clientExpect := setupConn(t, client)
defer client.Close()
// Send PUB via client connection
clientSend("PUB foo 2\r\nok\r\n")
clientSend("PING\r\n")
clientExpect(pongRe)
// We should only receive on one route, not both.
// Check both manually.
route1.SetReadDeadline(time.Now().Add(100 * time.Millisecond))
buf, _ := ioutil.ReadAll(route1)
route1.SetReadDeadline(time.Time{})
if len(buf) <= 0 {
route2.SetReadDeadline(time.Now().Add(100 * time.Millisecond))
buf, _ = ioutil.ReadAll(route2)
route2.SetReadDeadline(time.Time{})
if len(buf) <= 0 {
t.Fatal("Expected to get one message on a route, received none.")
}
}
matches := msgRe.FindAllSubmatch(buf, -1)
if len(matches) != 1 {
t.Fatalf("Expected 1 msg, got %d\n", len(matches))
}
checkMsg(t, matches[0], "foo", "", "", "2", "ok")
}
func TestRouteResendsLocalSubsOnReconnect(t *testing.T) {
s, opts := runRouteServer(t)
defer s.Shutdown()
client := createClientConn(t, opts.Host, opts.Port)
clientSend, clientExpect := setupConn(t, client)
// Setup a local subscription
clientSend("SUB foo 1\r\n")
clientSend("PING\r\n")
clientExpect(pongRe)
route := createRouteConn(t, opts.ClusterHost, opts.ClusterPort)
_, routeExpect := setupRouteEx(t, route, opts, "ROUTE:4222")
// Expect to see the local sub echoed through.
routeExpect(subRe)
// Close and re-open
route.Close()
route = createRouteConn(t, opts.ClusterHost, opts.ClusterPort)
_, routeExpect = setupRouteEx(t, route, opts, "ROUTE:4222")
// Expect to see the local sub echoed through.
routeExpect(subRe)
}

154
test/test.go
View File

@@ -3,12 +3,15 @@
package test
import (
"bytes"
"crypto/rand"
"encoding/hex"
"encoding/json"
"fmt"
"io"
"net"
"os/exec"
"regexp"
"runtime"
"strings"
"time"
@@ -46,7 +49,12 @@ func RunServer(opts *server.Options) *server.Server {
}
s := server.New(opts)
if s == nil {
panic("No nats server object returned.")
panic("No NATS Server object returned.")
}
// Start up clustering as well if needed.
if opts.ClusterPort != 0 {
s.StartRouting()
}
go s.AcceptLoop()
@@ -64,7 +72,7 @@ func RunServer(opts *server.Options) *server.Server {
conn.Close()
return s
}
panic("Unable to start NATs Server in Go Routine")
panic("Unable to start NATS Server in Go Routine")
}
func startServer(t tLogger, port int, other string) *natsServer {
@@ -104,11 +112,56 @@ func (s *natsServer) stopServer() {
}
}
func stackFatalf(t tLogger, f string, args ...interface{}) {
lines := make([]string, 0, 32)
msg := fmt.Sprintf(f, args...)
lines = append(lines, msg)
// Ignore ourselves
_, testFile, _, _ := runtime.Caller(0)
// Generate the Stack of callers:
for i := 0; true; i++ {
_, file, line, ok := runtime.Caller(i)
if ok == false {
break
}
if file == testFile {
continue
}
msg := fmt.Sprintf("%d - %s:%d", i, file, line)
lines = append(lines, msg)
}
t.Fatalf("%s", strings.Join(lines, "\n"))
}
func acceptRouteConn(t tLogger, host string, timeout time.Duration) net.Conn {
l, e := net.Listen("tcp", host)
if e != nil {
stackFatalf(t, "Error listening for route connection on %v: %v", host, e)
}
defer l.Close()
tl := l.(*net.TCPListener)
tl.SetDeadline(time.Now().Add(timeout))
conn, err := l.Accept()
if err != nil {
stackFatalf(t, "Did not receive a route connection request: %v", err)
}
return conn
}
func createRouteConn(t tLogger, host string, port int) net.Conn {
return createClientConn(t, host, port)
}
func createClientConn(t tLogger, host string, port int) net.Conn {
addr := fmt.Sprintf("%s:%d", host, port)
c, err := net.DialTimeout("tcp", addr, 1*time.Second)
if err != nil {
t.Fatalf("Could not connect to server: %v\n", err)
stackFatalf(t, "Could not connect to server: %v\n", err)
}
return c
}
@@ -119,7 +172,7 @@ func doConnect(t tLogger, c net.Conn, verbose, pedantic, ssl bool) {
var sinfo server.Info
err := json.Unmarshal(js, &sinfo)
if err != nil {
t.Fatalf("Could not unmarshal INFO json: %v\n", err)
stackFatalf(t, "Could not unmarshal INFO json: %v\n", err)
}
cs := fmt.Sprintf("CONNECT {\"verbose\":%v,\"pedantic\":%v,\"ssl_required\":%v}\r\n", verbose, pedantic, ssl)
sendProto(t, c, cs)
@@ -130,6 +183,46 @@ func doDefaultConnect(t tLogger, c net.Conn) {
doConnect(t, c, false, false, false)
}
func checkSocket(t tLogger, addr string, wait time.Duration) {
end := time.Now().Add(wait)
for time.Now().Before(end) {
conn, err := net.Dial("tcp", addr)
if err != nil {
time.Sleep(50 * time.Millisecond)
// Retry
continue
}
// We bound to the addr, so close and return success.
conn.Close()
return
}
// We have failed to bind the socket in the time allowed.
t.Fatalf("Failed to connect to the socket: %q", addr)
}
const CONNECT_F = "CONNECT {\"verbose\":false,\"user\":\"%s\",\"pass\":\"%s\",\"name\":\"%s\"}\r\n"
func doRouteAuthConnect(t tLogger, c net.Conn, user, pass, id string) {
cs := fmt.Sprintf(CONNECT_F, user, pass, id)
sendProto(t, c, cs)
}
func setupRouteEx(t tLogger, c net.Conn, opts *server.Options, id string) (sendFun, expectFun) {
user := opts.ClusterUsername
pass := opts.ClusterPassword
doRouteAuthConnect(t, c, user, pass, id)
send := sendCommand(t, c)
expect := expectCommand(t, c)
return send, expect
}
func setupRoute(t tLogger, c net.Conn, opts *server.Options) (sendFun, expectFun) {
u := make([]byte, 16)
io.ReadFull(rand.Reader, u)
id := fmt.Sprintf("ROUTER:%s", hex.EncodeToString(u))
return setupRouteEx(t, c, opts, id)
}
func setupConn(t tLogger, c net.Conn) (sendFun, expectFun) {
doDefaultConnect(t, c)
send := sendCommand(t, c)
@@ -158,20 +251,23 @@ func expectCommand(t tLogger, c net.Conn) expectFun {
func sendProto(t tLogger, c net.Conn, op string) {
n, err := c.Write([]byte(op))
if err != nil {
t.Fatalf("Error writing command to conn: %v\n", err)
stackFatalf(t, "Error writing command to conn: %v\n", err)
}
if n != len(op) {
t.Fatalf("Partial write: %d vs %d\n", n, len(op))
stackFatalf(t, "Partial write: %d vs %d\n", n, len(op))
}
}
var (
infoRe = regexp.MustCompile(`\AINFO\s+([^\r\n]+)\r\n`)
pingRe = regexp.MustCompile(`\APING\r\n`)
pongRe = regexp.MustCompile(`\APONG\r\n`)
msgRe = regexp.MustCompile(`(?:(?:MSG\s+([^\s]+)\s+([^\s]+)\s+(([^\s]+)[^\S\r\n]+)?(\d+)\r\n([^\\r\\n]*?)\r\n)+?)`)
okRe = regexp.MustCompile(`\A\+OK\r\n`)
errRe = regexp.MustCompile(`\A\-ERR\s+([^\r\n]+)\r\n`)
infoRe = regexp.MustCompile(`INFO\s+([^\r\n]+)\r\n`)
pingRe = regexp.MustCompile(`PING\r\n`)
pongRe = regexp.MustCompile(`PONG\r\n`)
msgRe = regexp.MustCompile(`(?:(?:MSG\s+([^\s]+)\s+([^\s]+)\s+(([^\s]+)[^\S\r\n]+)?(\d+)\s*\r\n([^\\r\\n]*?)\r\n)+?)`)
okRe = regexp.MustCompile(`\A\+OK\r\n`)
errRe = regexp.MustCompile(`\A\-ERR\s+([^\r\n]+)\r\n`)
subRe = regexp.MustCompile(`SUB\s+([^\s]+)((\s+)([^\s]+))?\s+([^\s]+)\r\n`)
unsubRe = regexp.MustCompile(`UNSUB\s+([^\s]+)(\s+(\d+))?\r\n`)
connectRe = regexp.MustCompile(`CONNECT\s+([^\r\n]+)\r\n`)
)
const (
@@ -183,44 +279,54 @@ const (
)
// Reuse expect buffer
// TODO(dlc) - This may be too simplistic in the long run, may need
// to consider holding onto data from previous reads matched by conn.
var expBuf = make([]byte, 32768)
// Test result from server against regexp
func expectResult(t tLogger, c net.Conn, re *regexp.Regexp) []byte {
// Wait for commands to be processed and results queued for read
// time.Sleep(10 * time.Millisecond)
c.SetReadDeadline(time.Now().Add(1 * time.Second))
defer c.SetReadDeadline(time.Time{})
n, err := c.Read(expBuf)
if n <= 0 && err != nil {
t.Fatalf("Error reading from conn: %v\n", err)
stackFatalf(t, "Error reading from conn: %v\n", err)
}
buf := expBuf[:n]
if !re.Match(buf) {
buf = bytes.Replace(buf, []byte("\r\n"), []byte("\\r\\n"), -1)
t.Fatalf("Response did not match expected: \n\tReceived:'%s'\n\tExpected:'%s'\n", buf, re)
stackFatalf(t, "Response did not match expected: \n\tReceived:'%q'\n\tExpected:'%s'\n", buf, re)
}
return buf
}
func expectNothing(t tLogger, c net.Conn) {
c.SetReadDeadline(time.Now().Add(100 * time.Millisecond))
defer c.SetReadDeadline(time.Time{})
n, err := c.Read(expBuf)
if err == nil && n > 0 {
stackFatalf(t, "Expected nothing, received: '%q'\n", expBuf[:n])
}
}
// This will check that we got what we expected.
func checkMsg(t tLogger, m [][]byte, subject, sid, reply, len, msg string) {
if string(m[SUB_INDEX]) != subject {
t.Fatalf("Did not get correct subject: expected '%s' got '%s'\n", subject, m[SUB_INDEX])
stackFatalf(t, "Did not get correct subject: expected '%s' got '%s'\n", subject, m[SUB_INDEX])
}
if string(m[SID_INDEX]) != sid {
t.Fatalf("Did not get correct sid: exepected '%s' got '%s'\n", sid, m[SID_INDEX])
if sid != "" && string(m[SID_INDEX]) != sid {
stackFatalf(t, "Did not get correct sid: expected '%s' got '%s'\n", sid, m[SID_INDEX])
}
if string(m[REPLY_INDEX]) != reply {
t.Fatalf("Did not get correct reply: exepected '%s' got '%s'\n", reply, m[REPLY_INDEX])
stackFatalf(t, "Did not get correct reply: expected '%s' got '%s'\n", reply, m[REPLY_INDEX])
}
if string(m[LEN_INDEX]) != len {
t.Fatalf("Did not get correct msg length: expected '%s' got '%s'\n", len, m[LEN_INDEX])
stackFatalf(t, "Did not get correct msg length: expected '%s' got '%s'\n", len, m[LEN_INDEX])
}
if string(m[MSG_INDEX]) != msg {
t.Fatalf("Did not get correct msg: expected '%s' got '%s'\n", msg, m[MSG_INDEX])
stackFatalf(t, "Did not get correct msg: expected '%s' got '%s'\n", msg, m[MSG_INDEX])
}
}
@@ -230,7 +336,7 @@ func expectMsgsCommand(t tLogger, ef expectFun) func(int) [][][]byte {
buf := ef(msgRe)
matches := msgRe.FindAllSubmatch(buf, -1)
if len(matches) != expected {
t.Fatalf("Did not get correct # msgs: %d vs %d\n", len(matches), expected)
stackFatalf(t, "Did not get correct # msgs: %d vs %d\n", len(matches), expected)
}
return matches
}