forked from nats-io/nats.go
-
Notifications
You must be signed in to change notification settings - Fork 0
/
nats.go
2979 lines (2597 loc) · 73.8 KB
/
nats.go
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
// Copyright 2012-2017 Apcera Inc. All rights reserved.
// A Go client for the NATS messaging system (https://nats.io).
package nats
import (
"bufio"
"bytes"
"crypto/tls"
"crypto/x509"
"encoding/json"
"errors"
"fmt"
"io/ioutil"
"math/rand"
"net"
"net/url"
"regexp"
"runtime"
"strconv"
"strings"
"sync"
"time"
"github.com/nats-io/go-nats/util"
"github.com/nats-io/nuid"
)
// Default Constants
const (
Version = "1.3.0"
DefaultURL = "nats://localhost:4222"
DefaultPort = 4222
DefaultMaxReconnect = 60
DefaultReconnectWait = 2 * time.Second
DefaultTimeout = 2 * time.Second
DefaultPingInterval = 2 * time.Minute
DefaultMaxPingOut = 2
DefaultMaxChanLen = 8192 // 8k
DefaultReconnectBufSize = 8 * 1024 * 1024 // 8MB
RequestChanLen = 8
LangString = "go"
)
// STALE_CONNECTION is for detection and proper handling of stale connections.
const STALE_CONNECTION = "stale connection"
// PERMISSIONS_ERR is for when nats server subject authorization has failed.
const PERMISSIONS_ERR = "permissions violation"
// AUTHORIZATION_ERR is for when nats server user authorization has failed.
const AUTHORIZATION_ERR = "authorization violation"
// Errors
var (
ErrConnectionClosed = errors.New("nats: connection closed")
ErrSecureConnRequired = errors.New("nats: secure connection required")
ErrSecureConnWanted = errors.New("nats: secure connection not available")
ErrBadSubscription = errors.New("nats: invalid subscription")
ErrTypeSubscription = errors.New("nats: invalid subscription type")
ErrBadSubject = errors.New("nats: invalid subject")
ErrSlowConsumer = errors.New("nats: slow consumer, messages dropped")
ErrTimeout = errors.New("nats: timeout")
ErrBadTimeout = errors.New("nats: timeout invalid")
ErrAuthorization = errors.New("nats: authorization violation")
ErrNoServers = errors.New("nats: no servers available for connection")
ErrJsonParse = errors.New("nats: connect message, json parse error")
ErrChanArg = errors.New("nats: argument needs to be a channel type")
ErrMaxPayload = errors.New("nats: maximum payload exceeded")
ErrMaxMessages = errors.New("nats: maximum messages delivered")
ErrSyncSubRequired = errors.New("nats: illegal call on an async subscription")
ErrMultipleTLSConfigs = errors.New("nats: multiple tls.Configs not allowed")
ErrNoInfoReceived = errors.New("nats: protocol exception, INFO not received")
ErrReconnectBufExceeded = errors.New("nats: outbound buffer limit exceeded")
ErrInvalidConnection = errors.New("nats: invalid connection")
ErrInvalidMsg = errors.New("nats: invalid message or message nil")
ErrInvalidArg = errors.New("nats: invalid argument")
ErrInvalidContext = errors.New("nats: invalid context")
ErrStaleConnection = errors.New("nats: " + STALE_CONNECTION)
)
// GetDefaultOptions returns default configuration options for the client.
func GetDefaultOptions() Options {
return Options{
AllowReconnect: true,
MaxReconnect: DefaultMaxReconnect,
ReconnectWait: DefaultReconnectWait,
Timeout: DefaultTimeout,
PingInterval: DefaultPingInterval,
MaxPingsOut: DefaultMaxPingOut,
SubChanLen: DefaultMaxChanLen,
ReconnectBufSize: DefaultReconnectBufSize,
}
}
// DEPRECATED: Use GetDefaultOptions() instead.
// DefaultOptions is not safe for use by multiple clients.
// For details see #308.
var DefaultOptions = GetDefaultOptions()
// Status represents the state of the connection.
type Status int
const (
DISCONNECTED = Status(iota)
CONNECTED
CLOSED
RECONNECTING
CONNECTING
)
// ConnHandler is used for asynchronous events such as
// disconnected and closed connections.
type ConnHandler func(*Conn)
// ErrHandler is used to process asynchronous errors encountered
// while processing inbound messages.
type ErrHandler func(*Conn, *Subscription, error)
// asyncCB is used to preserve order for async callbacks.
type asyncCB func()
// Option is a function on the options for a connection.
type Option func(*Options) error
// Options can be used to create a customized connection.
type Options struct {
// Url represents a single NATS server url to which the client
// will be connecting. If the Servers option is also set, it
// then becomes the first server in the Servers array.
Url string
// Servers is a configured set of servers which this client
// will use when attempting to connect.
Servers []string
// NoRandomize configures whether we will randomize the
// server pool.
NoRandomize bool
// Name is an optional name label which will be sent to the server
// on CONNECT to identify the client.
Name string
// Verbose signals the server to send an OK ack for commands
// successfully processed by the server.
Verbose bool
// Pedantic signals the server whether it should be doing further
// validation of subjects.
Pedantic bool
// Secure enables TLS secure connections that skip server
// verification by default. NOT RECOMMENDED.
Secure bool
// TLSConfig is a custom TLS configuration to use for secure
// transports.
TLSConfig *tls.Config
// AllowReconnect enables reconnection logic to be used when we
// encounter a disconnect from the current server.
AllowReconnect bool
// MaxReconnect sets the number of reconnect attempts that will be
// tried before giving up. If negative, then it will never give up
// trying to reconnect.
MaxReconnect int
// ReconnectWait sets the time to backoff after attempting a reconnect
// to a server that we were already connected to previously.
ReconnectWait time.Duration
// Timeout sets the timeout for a Dial operation on a connection.
Timeout time.Duration
// FlusherTimeout is the maximum time to wait for the flusher loop
// to be able to finish writing to the underlying connection.
FlusherTimeout time.Duration
// PingInterval is the period at which the client will be sending ping
// commands to the server, disabled if 0 or negative.
PingInterval time.Duration
// MaxPingsOut is the maximum number of pending ping commands that can
// be awaiting a response before raising an ErrStaleConnection error.
MaxPingsOut int
// ClosedCB sets the closed handler that is called when a client will
// no longer be connected.
ClosedCB ConnHandler
// DisconnectedCB sets the disconnected handler that is called
// whenever the connection is disconnected.
DisconnectedCB ConnHandler
// ReconnectedCB sets the reconnected handler called whenever
// the connection is successfully reconnected.
ReconnectedCB ConnHandler
// DiscoveredServersCB sets the callback that is invoked whenever a new
// server has joined the cluster.
DiscoveredServersCB ConnHandler
// AsyncErrorCB sets the async error handler (e.g. slow consumer errors)
AsyncErrorCB ErrHandler
// ReconnectBufSize is the size of the backing bufio during reconnect.
// Once this has been exhausted publish operations will return an error.
ReconnectBufSize int
// SubChanLen is the size of the buffered channel used between the socket
// Go routine and the message delivery for SyncSubscriptions.
// NOTE: This does not affect AsyncSubscriptions which are
// dictated by PendingLimits()
SubChanLen int
// User sets the username to be used when connecting to the server.
User string
// Password sets the password to be used when connecting to a server.
Password string
// Token sets the token to be used when connecting to a server.
Token string
// Dialer allows a custom Dialer when forming connections.
Dialer *net.Dialer
// UseOldRequestStyle forces the old method of Requests that utilize
// a new Inbox and a new Subscription for each request.
UseOldRequestStyle bool
}
const (
// Scratch storage for assembling protocol headers
scratchSize = 512
// The size of the bufio reader/writer on top of the socket.
defaultBufSize = 32768
// The buffered size of the flush "kick" channel
flushChanSize = 1024
// Default server pool size
srvPoolSize = 4
// Channel size for the async callback handler.
asyncCBChanSize = 32
// NUID size
nuidSize = 22
)
// A Conn represents a bare connection to a nats-server.
// It can send and receive []byte payloads.
type Conn struct {
// Keep all members for which we use atomic at the beginning of the
// struct and make sure they are all 64bits (or use padding if necessary).
// atomic.* functions crash on 32bit machines if operand is not aligned
// at 64bit. See https://github.com/golang/go/issues/599
Statistics
mu sync.Mutex
Opts Options
wg *sync.WaitGroup
url *url.URL
conn net.Conn
srvPool []*srv
urls map[string]struct{} // Keep track of all known URLs (used by processInfo)
bw *bufio.Writer
pending *bytes.Buffer
fch chan struct{}
info serverInfo
ssid int64
subsMu sync.RWMutex
subs map[int64]*Subscription
mch chan *Msg
ach chan asyncCB
pongs []chan struct{}
scratch [scratchSize]byte
status Status
initc bool // true if the connection is performing the initial connect
err error
ps *parseState
ptmr *time.Timer
pout int
// New style response handler
respSub string // The wildcard subject
respMux *Subscription // A single response subscription
respMap map[string]chan *Msg // Request map for the response msg channels
respSetup sync.Once // Ensures response subscription occurs once
}
// A Subscription represents interest in a given subject.
type Subscription struct {
mu sync.Mutex
sid int64
// Subject that represents this subscription. This can be different
// than the received subject inside a Msg if this is a wildcard.
Subject string
// Optional queue group name. If present, all subscriptions with the
// same name will form a distributed queue, and each message will
// only be processed by one member of the group.
Queue string
delivered uint64
max uint64
conn *Conn
mcb MsgHandler
mch chan *Msg
closed bool
sc bool
connClosed bool
// Type of Subscription
typ SubscriptionType
// Async linked list
pHead *Msg
pTail *Msg
pCond *sync.Cond
// Pending stats, async subscriptions, high-speed etc.
pMsgs int
pBytes int
pMsgsMax int
pBytesMax int
pMsgsLimit int
pBytesLimit int
dropped int
}
// Msg is a structure used by Subscribers and PublishMsg().
type Msg struct {
Subject string
Reply string
Data []byte
Sub *Subscription
next *Msg
}
// Tracks various stats received and sent on this connection,
// including counts for messages and bytes.
type Statistics struct {
InMsgs uint64
OutMsgs uint64
InBytes uint64
OutBytes uint64
Reconnects uint64
}
// Tracks individual backend servers.
type srv struct {
url *url.URL
didConnect bool
reconnects int
lastAttempt time.Time
isImplicit bool
}
type serverInfo struct {
Id string `json:"server_id"`
Host string `json:"host"`
Port uint `json:"port"`
Version string `json:"version"`
AuthRequired bool `json:"auth_required"`
TLSRequired bool `json:"tls_required"`
MaxPayload int64 `json:"max_payload"`
ConnectURLs []string `json:"connect_urls,omitempty"`
}
const (
// clientProtoZero is the original client protocol from 2009.
// http://nats.io/documentation/internals/nats-protocol/
clientProtoZero = iota
// clientProtoInfo signals a client can receive more then the original INFO block.
// This can be used to update clients on other cluster members, etc.
clientProtoInfo
)
type connectInfo struct {
Verbose bool `json:"verbose"`
Pedantic bool `json:"pedantic"`
User string `json:"user,omitempty"`
Pass string `json:"pass,omitempty"`
Token string `json:"auth_token,omitempty"`
TLS bool `json:"tls_required"`
Name string `json:"name"`
Lang string `json:"lang"`
Version string `json:"version"`
Protocol int `json:"protocol"`
}
// MsgHandler is a callback function that processes messages delivered to
// asynchronous subscribers.
type MsgHandler func(msg *Msg)
// Connect will attempt to connect to the NATS system.
// The url can contain username/password semantics. e.g. nats://derek:pass@localhost:4222
// Comma separated arrays are also supported, e.g. urlA, urlB.
// Options start with the defaults but can be overridden.
func Connect(url string, options ...Option) (*Conn, error) {
opts := GetDefaultOptions()
opts.Servers = processUrlString(url)
for _, opt := range options {
if err := opt(&opts); err != nil {
return nil, err
}
}
return opts.Connect()
}
// Options that can be passed to Connect.
// Name is an Option to set the client name.
func Name(name string) Option {
return func(o *Options) error {
o.Name = name
return nil
}
}
// Secure is an Option to enable TLS secure connections that skip server verification by default.
// Pass a TLS Configuration for proper TLS.
func Secure(tls ...*tls.Config) Option {
return func(o *Options) error {
o.Secure = true
// Use of variadic just simplifies testing scenarios. We only take the first one.
// fixme(DLC) - Could panic if more than one. Could also do TLS option.
if len(tls) > 1 {
return ErrMultipleTLSConfigs
}
if len(tls) == 1 {
o.TLSConfig = tls[0]
}
return nil
}
}
// RootCAs is a helper option to provide the RootCAs pool from a list of filenames. If Secure is
// not already set this will set it as well.
func RootCAs(file ...string) Option {
return func(o *Options) error {
pool := x509.NewCertPool()
for _, f := range file {
rootPEM, err := ioutil.ReadFile(f)
if err != nil || rootPEM == nil {
return fmt.Errorf("nats: error loading or parsing rootCA file: %v", err)
}
ok := pool.AppendCertsFromPEM([]byte(rootPEM))
if !ok {
return fmt.Errorf("nats: failed to parse root certificate from %q", f)
}
}
if o.TLSConfig == nil {
o.TLSConfig = &tls.Config{MinVersion: tls.VersionTLS12}
}
o.TLSConfig.RootCAs = pool
o.Secure = true
return nil
}
}
// ClientCert is a helper option to provide the client certificate from a file. If Secure is
// not already set this will set it as well
func ClientCert(certFile, keyFile string) Option {
return func(o *Options) error {
cert, err := tls.LoadX509KeyPair(certFile, keyFile)
if err != nil {
return fmt.Errorf("nats: error loading client certificate: %v", err)
}
cert.Leaf, err = x509.ParseCertificate(cert.Certificate[0])
if err != nil {
return fmt.Errorf("nats: error parsing client certificate: %v", err)
}
if o.TLSConfig == nil {
o.TLSConfig = &tls.Config{MinVersion: tls.VersionTLS12}
}
o.TLSConfig.Certificates = []tls.Certificate{cert}
o.Secure = true
return nil
}
}
// NoReconnect is an Option to turn off reconnect behavior.
func NoReconnect() Option {
return func(o *Options) error {
o.AllowReconnect = false
return nil
}
}
// DontRandomize is an Option to turn off randomizing the server pool.
func DontRandomize() Option {
return func(o *Options) error {
o.NoRandomize = true
return nil
}
}
// ReconnectWait is an Option to set the wait time between reconnect attempts.
func ReconnectWait(t time.Duration) Option {
return func(o *Options) error {
o.ReconnectWait = t
return nil
}
}
// MaxReconnects is an Option to set the maximum number of reconnect attempts.
func MaxReconnects(max int) Option {
return func(o *Options) error {
o.MaxReconnect = max
return nil
}
}
// Timeout is an Option to set the timeout for Dial on a connection.
func Timeout(t time.Duration) Option {
return func(o *Options) error {
o.Timeout = t
return nil
}
}
// DisconnectHandler is an Option to set the disconnected handler.
func DisconnectHandler(cb ConnHandler) Option {
return func(o *Options) error {
o.DisconnectedCB = cb
return nil
}
}
// ReconnectHandler is an Option to set the reconnected handler.
func ReconnectHandler(cb ConnHandler) Option {
return func(o *Options) error {
o.ReconnectedCB = cb
return nil
}
}
// ClosedHandler is an Option to set the closed handler.
func ClosedHandler(cb ConnHandler) Option {
return func(o *Options) error {
o.ClosedCB = cb
return nil
}
}
// DiscoveredServersHandler is an Option to set the new servers handler.
func DiscoveredServersHandler(cb ConnHandler) Option {
return func(o *Options) error {
o.DiscoveredServersCB = cb
return nil
}
}
// ErrHandler is an Option to set the async error handler.
func ErrorHandler(cb ErrHandler) Option {
return func(o *Options) error {
o.AsyncErrorCB = cb
return nil
}
}
// UserInfo is an Option to set the username and password to
// use when not included directly in the URLs.
func UserInfo(user, password string) Option {
return func(o *Options) error {
o.User = user
o.Password = password
return nil
}
}
// Token is an Option to set the token to use when not included
// directly in the URLs.
func Token(token string) Option {
return func(o *Options) error {
o.Token = token
return nil
}
}
// Dialer is an Option to set the dialer which will be used when
// attempting to establish a connection.
func Dialer(dialer *net.Dialer) Option {
return func(o *Options) error {
o.Dialer = dialer
return nil
}
}
// UseOldRequestyStyle is an Option to force usage of the old Request style.
func UseOldRequestStyle() Option {
return func(o *Options) error {
o.UseOldRequestStyle = true
return nil
}
}
// Handler processing
// SetDisconnectHandler will set the disconnect event handler.
func (nc *Conn) SetDisconnectHandler(dcb ConnHandler) {
if nc == nil {
return
}
nc.mu.Lock()
defer nc.mu.Unlock()
nc.Opts.DisconnectedCB = dcb
}
// SetReconnectHandler will set the reconnect event handler.
func (nc *Conn) SetReconnectHandler(rcb ConnHandler) {
if nc == nil {
return
}
nc.mu.Lock()
defer nc.mu.Unlock()
nc.Opts.ReconnectedCB = rcb
}
// SetDiscoveredServersHandler will set the discovered servers handler.
func (nc *Conn) SetDiscoveredServersHandler(dscb ConnHandler) {
if nc == nil {
return
}
nc.mu.Lock()
defer nc.mu.Unlock()
nc.Opts.DiscoveredServersCB = dscb
}
// SetClosedHandler will set the reconnect event handler.
func (nc *Conn) SetClosedHandler(cb ConnHandler) {
if nc == nil {
return
}
nc.mu.Lock()
defer nc.mu.Unlock()
nc.Opts.ClosedCB = cb
}
// SetErrHandler will set the async error handler.
func (nc *Conn) SetErrorHandler(cb ErrHandler) {
if nc == nil {
return
}
nc.mu.Lock()
defer nc.mu.Unlock()
nc.Opts.AsyncErrorCB = cb
}
// Process the url string argument to Connect. Return an array of
// urls, even if only one.
func processUrlString(url string) []string {
urls := strings.Split(url, ",")
for i, s := range urls {
urls[i] = strings.TrimSpace(s)
}
return urls
}
// Connect will attempt to connect to a NATS server with multiple options.
func (o Options) Connect() (*Conn, error) {
nc := &Conn{Opts: o}
// Some default options processing.
if nc.Opts.MaxPingsOut == 0 {
nc.Opts.MaxPingsOut = DefaultMaxPingOut
}
// Allow old default for channel length to work correctly.
if nc.Opts.SubChanLen == 0 {
nc.Opts.SubChanLen = DefaultMaxChanLen
}
// Default ReconnectBufSize
if nc.Opts.ReconnectBufSize == 0 {
nc.Opts.ReconnectBufSize = DefaultReconnectBufSize
}
// Ensure that Timeout is not 0
if nc.Opts.Timeout == 0 {
nc.Opts.Timeout = DefaultTimeout
}
// Allow custom Dialer for connecting using DialTimeout by default
if nc.Opts.Dialer == nil {
nc.Opts.Dialer = &net.Dialer{
Timeout: nc.Opts.Timeout,
}
}
if err := nc.setupServerPool(); err != nil {
return nil, err
}
// Create the async callback channel.
nc.ach = make(chan asyncCB, asyncCBChanSize)
if err := nc.connect(); err != nil {
return nil, err
}
// Spin up the async cb dispatcher on success
go nc.asyncDispatch()
return nc, nil
}
const (
_CRLF_ = "\r\n"
_EMPTY_ = ""
_SPC_ = " "
_PUB_P_ = "PUB "
)
const (
_OK_OP_ = "+OK"
_ERR_OP_ = "-ERR"
_MSG_OP_ = "MSG"
_PING_OP_ = "PING"
_PONG_OP_ = "PONG"
_INFO_OP_ = "INFO"
)
const (
conProto = "CONNECT %s" + _CRLF_
pingProto = "PING" + _CRLF_
pongProto = "PONG" + _CRLF_
pubProto = "PUB %s %s %d" + _CRLF_
subProto = "SUB %s %s %d" + _CRLF_
unsubProto = "UNSUB %d %s" + _CRLF_
okProto = _OK_OP_ + _CRLF_
)
// Return the currently selected server
func (nc *Conn) currentServer() (int, *srv) {
for i, s := range nc.srvPool {
if s == nil {
continue
}
if s.url == nc.url {
return i, s
}
}
return -1, nil
}
// Pop the current server and put onto the end of the list. Select head of list as long
// as number of reconnect attempts under MaxReconnect.
func (nc *Conn) selectNextServer() (*srv, error) {
i, s := nc.currentServer()
if i < 0 {
return nil, ErrNoServers
}
sp := nc.srvPool
num := len(sp)
copy(sp[i:num-1], sp[i+1:num])
maxReconnect := nc.Opts.MaxReconnect
if maxReconnect < 0 || s.reconnects < maxReconnect {
nc.srvPool[num-1] = s
} else {
nc.srvPool = sp[0 : num-1]
}
if len(nc.srvPool) <= 0 {
nc.url = nil
return nil, ErrNoServers
}
nc.url = nc.srvPool[0].url
return nc.srvPool[0], nil
}
// Will assign the correct server to the nc.Url
func (nc *Conn) pickServer() error {
nc.url = nil
if len(nc.srvPool) <= 0 {
return ErrNoServers
}
for _, s := range nc.srvPool {
if s != nil {
nc.url = s.url
return nil
}
}
return ErrNoServers
}
const tlsScheme = "tls"
// Create the server pool using the options given.
// We will place a Url option first, followed by any
// Server Options. We will randomize the server pool unlesss
// the NoRandomize flag is set.
func (nc *Conn) setupServerPool() error {
nc.srvPool = make([]*srv, 0, srvPoolSize)
nc.urls = make(map[string]struct{}, srvPoolSize)
// Create srv objects from each url string in nc.Opts.Servers
// and add them to the pool
for _, urlString := range nc.Opts.Servers {
if err := nc.addURLToPool(urlString, false); err != nil {
return err
}
}
// Randomize if allowed to
if !nc.Opts.NoRandomize {
nc.shufflePool()
}
// Normally, if this one is set, Options.Servers should not be,
// but we always allowed that, so continue to do so.
if nc.Opts.Url != _EMPTY_ {
// Add to the end of the array
if err := nc.addURLToPool(nc.Opts.Url, false); err != nil {
return err
}
// Then swap it with first to guarantee that Options.Url is tried first.
last := len(nc.srvPool) - 1
if last > 0 {
nc.srvPool[0], nc.srvPool[last] = nc.srvPool[last], nc.srvPool[0]
}
} else if len(nc.srvPool) <= 0 {
// Place default URL if pool is empty.
if err := nc.addURLToPool(DefaultURL, false); err != nil {
return err
}
}
// Check for Scheme hint to move to TLS mode.
for _, srv := range nc.srvPool {
if srv.url.Scheme == tlsScheme {
// FIXME(dlc), this is for all in the pool, should be case by case.
nc.Opts.Secure = true
if nc.Opts.TLSConfig == nil {
nc.Opts.TLSConfig = &tls.Config{MinVersion: tls.VersionTLS12}
}
}
}
return nc.pickServer()
}
// addURLToPool adds an entry to the server pool
func (nc *Conn) addURLToPool(sURL string, implicit bool) error {
u, err := url.Parse(sURL)
if err != nil {
return err
}
s := &srv{url: u, isImplicit: implicit}
nc.srvPool = append(nc.srvPool, s)
nc.urls[u.Host] = struct{}{}
return nil
}
// shufflePool swaps randomly elements in the server pool
func (nc *Conn) shufflePool() {
if len(nc.srvPool) <= 1 {
return
}
source := rand.NewSource(time.Now().UnixNano())
r := rand.New(source)
for i := range nc.srvPool {
j := r.Intn(i + 1)
nc.srvPool[i], nc.srvPool[j] = nc.srvPool[j], nc.srvPool[i]
}
}
// createConn will connect to the server and wrap the appropriate
// bufio structures. It will do the right thing when an existing
// connection is in place.
func (nc *Conn) createConn() (err error) {
if nc.Opts.Timeout < 0 {
return ErrBadTimeout
}
if _, cur := nc.currentServer(); cur == nil {
return ErrNoServers
} else {
cur.lastAttempt = time.Now()
}
dialer := nc.Opts.Dialer
nc.conn, err = dialer.Dial("tcp", nc.url.Host)
if err != nil {
return err
}
// No clue why, but this stalls and kills performance on Mac (Mavericks).
// https://code.google.com/p/go/issues/detail?id=6930
//if ip, ok := nc.conn.(*net.TCPConn); ok {
// ip.SetReadBuffer(defaultBufSize)
//}
if nc.pending != nil && nc.bw != nil {
// Move to pending buffer.
nc.bw.Flush()
}
nc.bw = bufio.NewWriterSize(nc.conn, defaultBufSize)
return nil
}
// makeTLSConn will wrap an existing Conn using TLS
func (nc *Conn) makeTLSConn() {
// Allow the user to configure their own tls.Config structure, otherwise
// default to InsecureSkipVerify.
// TODO(dlc) - We should make the more secure version the default.
if nc.Opts.TLSConfig != nil {
tlsCopy := util.CloneTLSConfig(nc.Opts.TLSConfig)
// If its blank we will override it with the current host
if tlsCopy.ServerName == _EMPTY_ {
h, _, _ := net.SplitHostPort(nc.url.Host)
tlsCopy.ServerName = h
}
nc.conn = tls.Client(nc.conn, tlsCopy)
} else {
nc.conn = tls.Client(nc.conn, &tls.Config{InsecureSkipVerify: true})
}
conn := nc.conn.(*tls.Conn)
conn.Handshake()
nc.bw = bufio.NewWriterSize(nc.conn, defaultBufSize)
}
// waitForExits will wait for all socket watcher Go routines to
// be shutdown before proceeding.
func (nc *Conn) waitForExits(wg *sync.WaitGroup) {
// Kick old flusher forcefully.
select {
case nc.fch <- struct{}{}:
default:
}
// Wait for any previous go routines.
if wg != nil {
wg.Wait()
}
}
// spinUpGoRoutines will launch the Go routines responsible for
// reading and writing to the socket. This will be launched via a
// go routine itself to release any locks that may be held.
// We also use a WaitGroup to make sure we only start them on a
// reconnect when the previous ones have exited.
func (nc *Conn) spinUpGoRoutines() {
// Make sure everything has exited.
nc.waitForExits(nc.wg)
// Create a new waitGroup instance for this run.
nc.wg = &sync.WaitGroup{}
// We will wait on both.
nc.wg.Add(2)
// Spin up the readLoop and the socket flusher.
go nc.readLoop(nc.wg)
go nc.flusher(nc.wg)
nc.mu.Lock()
if nc.Opts.PingInterval > 0 {
if nc.ptmr == nil {
nc.ptmr = time.AfterFunc(nc.Opts.PingInterval, nc.processPingTimer)
} else {
nc.ptmr.Reset(nc.Opts.PingInterval)
}
}
nc.mu.Unlock()
}
// Report the connected server's Url
func (nc *Conn) ConnectedUrl() string {
if nc == nil {
return _EMPTY_
}
nc.mu.Lock()
defer nc.mu.Unlock()
if nc.status != CONNECTED {
return _EMPTY_
}
return nc.url.String()
}
// Report the connected server's Id
func (nc *Conn) ConnectedServerId() string {
if nc == nil {
return _EMPTY_
}
nc.mu.Lock()
defer nc.mu.Unlock()
if nc.status != CONNECTED {
return _EMPTY_
}
return nc.info.Id
}
// Low level setup for structs, etc
func (nc *Conn) setup() {
nc.subs = make(map[int64]*Subscription)
nc.pongs = make([]chan struct{}, 0, 8)
nc.fch = make(chan struct{}, flushChanSize)