-
Notifications
You must be signed in to change notification settings - Fork 29
/
Copy pathmulticlient.go
1120 lines (970 loc) · 33.8 KB
/
multiclient.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
package nkn
import (
"context"
"errors"
"log"
"net"
"regexp"
"sort"
"strconv"
"strings"
"sync"
"time"
"github.com/hashicorp/go-multierror"
"github.com/nknorg/ncp-go"
"github.com/nknorg/nkn-sdk-go/payloads"
"github.com/nknorg/nkn/v2/transaction"
"github.com/nknorg/nkn/v2/util/address"
"github.com/nknorg/nkngomobile"
"github.com/patrickmn/go-cache"
)
const (
// MultiClientIdentifierRe is the regular expression to check whether an
// identifier is a multiclient protocol identifier.
MultiClientIdentifierRe = "^__\\d+__$"
// DefaultSessionAllowAddr is the default session allow address if none is
// provided when calling listen.
DefaultSessionAllowAddr = ".*"
// SessionIDSize is the default session id size in bytes.
SessionIDSize = MessageIDSize
// Channel length for session that is received but not accepted by user.
acceptSessionBufSize = 128
)
var (
multiClientIdentifierRe = regexp.MustCompile(MultiClientIdentifierRe)
)
func addMultiClientPrefix(dest []string, clientID int) []string {
result := make([]string, len(dest))
for i, addr := range dest {
result[i] = addIdentifier(addr, clientID)
}
return result
}
func addIdentifier(addr string, id int) string {
if id < 0 {
return addr
}
return addIdentifierPrefix(addr, "__"+strconv.Itoa(id)+"__")
}
func removeIdentifier(src string) (string, string) {
s := strings.SplitN(src, ".", 2)
if len(s) > 1 {
if multiClientIdentifierRe.MatchString(s[0]) {
return s[1], s[0]
}
}
return src, ""
}
// MultiClient sends and receives data using multiple NKN clients concurrently
// to improve reliability and latency. In addition, it supports session mode, a
// reliable streaming protocol similar to TCP based on ncp
// (https://github.com/nknorg/ncp-go).
type MultiClient struct {
OnConnect *OnConnect // Event emitting channel when at least one client connects to node and becomes ready to send messages. One should only use the first event of the channel.
OnMessage *OnMessage // Event emitting channel when at least one client receives a message (not including reply or ACK).
config *ClientConfig
offset int
addr *ClientAddr
acceptSession chan *ncp.Session
onClose chan struct{}
msgCache *cache.Cache
resolvers []Resolver
lock sync.RWMutex
clients map[int]*Client
defaultClient *Client
acceptAddrs []*regexp.Regexp
isClosed bool
createDone bool
sessionLock sync.Mutex
sessions map[string]*ncp.Session
}
// NewMultiClientV2 creates a MultiClient with an account, an optional identifier,
// and an optional client config. For any zero value field in config, the default
// client config value will be used. If config is nil, the default client config
// will be used.
func NewMultiClientV2(account *Account, identifier string, config *ClientConfig) (*MultiClient, error) {
config, err := MergeClientConfig(config)
if err != nil {
return nil, err
}
return NewMultiClient(account, identifier, config.MultiClientNumClients, config.MultiClientOriginalClient, config)
}
// NewMultiClient creates a multiclient with an account, an optional identifier,
// number of sub clients to create, whether to create original client without
// identifier prefix, and an optional client config that will be applied to all
// clients created. For any zero value field in config, the default client
// config value will be used. If config is nil, the default client config will
// be used.
//
// Deprecated: Use NewMultiClientV2 instead.
func NewMultiClient(account *Account, baseIdentifier string, numSubClients int, originalClient bool, config *ClientConfig) (*MultiClient, error) {
config, err := MergeClientConfig(config)
if err != nil {
return nil, err
}
numClients := numSubClients
offset := 0
if originalClient {
numClients++
offset = 1
}
addr := address.MakeAddressString(account.PublicKey, baseIdentifier)
gomobileResolvers := config.Resolvers.Elems()
resolvers := make([]Resolver, 0, len(gomobileResolvers))
for i := 0; i < len(resolvers); i++ {
r, ok := gomobileResolvers[i].(Resolver)
if !ok {
return nil, ErrInvalidResolver
}
resolvers = append(resolvers, r)
}
m := &MultiClient{
config: config,
offset: offset,
addr: NewClientAddr(addr),
OnConnect: NewOnConnect(1, nil),
OnMessage: NewOnMessage(int(config.MsgChanLen), nil),
acceptSession: make(chan *ncp.Session, acceptSessionBufSize),
onClose: make(chan struct{}),
msgCache: cache.New(time.Duration(config.MsgCacheExpiration)*time.Millisecond, time.Duration(config.MsgCacheCleanupInterval)*time.Millisecond),
clients: make(map[int]*Client, numClients),
defaultClient: nil,
sessions: make(map[string]*ncp.Session),
resolvers: resolvers,
}
var wg sync.WaitGroup
defaultClientIdx := numSubClients
success := make(chan struct{}, 1)
fail := make(chan struct{}, 1)
for i := -offset; i < numSubClients; i++ {
wg.Add(1)
go func(i int) {
client, err := NewClient(account, addIdentifier(baseIdentifier, i), config)
if err != nil {
log.Println(err)
wg.Done()
return
}
m.lock.Lock()
if m.isClosed {
err := client.Close()
if err != nil {
log.Println(err)
}
wg.Done()
m.lock.Unlock()
return
}
m.clients[i] = client
if i < defaultClientIdx {
m.defaultClient = client
defaultClientIdx = i
}
m.lock.Unlock()
select {
case success <- struct{}{}:
default:
}
wg.Done()
go func() {
for {
select {
case node, ok := <-client.OnConnect.C:
if !ok {
return
}
m.lock.RLock()
if m.isClosed {
m.lock.RUnlock()
return
}
m.OnConnect.receive(node)
m.lock.RUnlock()
case <-m.onClose:
return
}
}
}()
for {
select {
case msg, ok := <-client.OnMessage.C:
if !ok {
return
}
if msg.Type == SessionType {
if !msg.Encrypted {
continue
}
err := m.handleSessionMsg(addIdentifier("", i-offset), msg.Src, msg.MessageID, msg.Data)
if err != nil {
if !errors.Is(err, ncp.ErrSessionClosed) && !errors.Is(err, ErrAddrNotAllowed) {
log.Println(err)
}
continue
}
} else {
cacheKey := string(msg.MessageID)
if _, ok := m.msgCache.Get(cacheKey); ok {
continue
}
m.msgCache.Set(cacheKey, nil, cache.DefaultExpiration)
msg.Src, _ = removeIdentifier(msg.Src)
if msg.NoReply {
msg.reply = func(data interface{}) error {
return nil
}
} else {
msg.reply = func(data interface{}) error {
payload, err := NewReplyPayload(data, msg.MessageID)
if err != nil {
return err
}
if err := m.send([]string{msg.Src}, payload, msg.Encrypted, 0); err != nil {
return err
}
return nil
}
}
m.lock.RLock()
if m.isClosed {
m.lock.RUnlock()
return
}
m.OnMessage.receive(msg, true)
m.lock.RUnlock()
}
case <-m.onClose:
return
}
}
}(i)
}
go func() {
wg.Wait()
m.lock.Lock()
m.createDone = true
m.lock.Unlock()
select {
case fail <- struct{}{}:
default:
}
}()
select {
case <-success:
return m, nil
case <-fail:
return nil, ErrCreateClientFailed
}
}
// Account returns the account of the multiclient.
func (m *MultiClient) Account() *Account {
return m.GetDefaultClient().account
}
// Seed returns the secret seed of the multiclient. Secret seed can be used to
// create client/wallet with the same key pair and should be kept secret and
// safe.
func (m *MultiClient) Seed() []byte {
return m.GetDefaultClient().Seed()
}
// PubKey returns the public key of the multiclient.
func (m *MultiClient) PubKey() []byte {
return m.GetDefaultClient().PubKey()
}
// Address returns the NKN client address of the multiclient. Client address is
// in the form of
//
// identifier.pubKeyHex
//
// if identifier is not an empty string, or
//
// pubKeyHex
//
// if identifier is an empty string.
//
// Note that client address is different from wallet address using the same key
// pair (account). Wallet address can be computed from client address, but NOT
// vice versa.
func (m *MultiClient) Address() string {
return m.addr.String()
}
// Addr returns the NKN client address of the multiclient as net.Addr interface,
// with Network() returns "nkn" and String() returns the same value as
// multiclient.Address().
func (m *MultiClient) Addr() net.Addr {
return m.addr
}
// GetClients returns all clients of the multiclient with client index as key.
// Subclients index starts from 0, and original client (if created) has index
// -1.
func (m *MultiClient) GetClients() map[int]*Client {
m.lock.RLock()
defer m.lock.RUnlock()
if m.createDone {
return m.clients
}
clients := make(map[int]*Client, len(m.clients))
for i, client := range m.clients {
clients[i] = client
}
return clients
}
// GetClient returns a client with a given index.
func (m *MultiClient) GetClient(i int) *Client {
m.lock.RLock()
defer m.lock.RUnlock()
return m.clients[i]
}
// GetDefaultClient returns the default client, which is the client with
// the smallest index.
func (m *MultiClient) GetDefaultClient() *Client {
m.lock.RLock()
defer m.lock.RUnlock()
return m.defaultClient
}
// SendWithClient sends bytes or string data to one or multiple destinations
// using a specific client with given index.
func (m *MultiClient) SendWithClient(clientID int, dests *nkngomobile.StringArray, data interface{}, config *MessageConfig) (*OnMessage, error) {
client := m.GetClient(clientID)
if client == nil {
return nil, ErrNilClient
}
config, err := MergeMessageConfig(m.config.MessageConfig, config)
if err != nil {
return nil, err
}
ctx, cancel := context.WithTimeout(context.Background(), time.Duration(m.config.ResolverTimeout)*time.Millisecond)
defer cancel()
destArr, err := m.ResolveDestsContext(ctx, dests)
if err != nil {
return nil, err
}
payload, err := newMessagePayload(data, config.MessageID, config.NoReply)
if err != nil {
return nil, err
}
if err := m.sendWithClient(clientID, destArr.Elems(), payload, !config.Unencrypted, config.MaxHoldingSeconds); err != nil {
return nil, err
}
onReply := NewOnMessage(1, nil)
if !config.NoReply {
client.responseChannels.Add(string(payload.MessageId), onReply, cache.DefaultExpiration)
}
return onReply, nil
}
// SendBinaryWithClient is a wrapper of SendWithClient without interface type
// for gomobile compatibility.
func (m *MultiClient) SendBinaryWithClient(clientID int, dests *nkngomobile.StringArray, data []byte, config *MessageConfig) (*OnMessage, error) {
return m.SendWithClient(clientID, dests, data, config)
}
// SendTextWithClient is a wrapper of SendWithClient without interface type for
// gomobile compatibility.
func (m *MultiClient) SendTextWithClient(clientID int, dests *nkngomobile.StringArray, data string, config *MessageConfig) (*OnMessage, error) {
return m.SendWithClient(clientID, dests, data, config)
}
func (m *MultiClient) sendWithClient(clientID int, dests []string, payload *payloads.Payload, encrypted bool, maxHoldingSeconds int32) error {
client := m.GetClient(clientID)
if client == nil {
return ErrNilClient
}
return client.send(addMultiClientPrefix(dests, clientID), payload, encrypted, maxHoldingSeconds)
}
// Send sends bytes or string data to one or multiple destinations with an
// optional config. Returned OnMessage channel will emit if a reply or ACK for
// this message is received.
func (m *MultiClient) Send(dests *nkngomobile.StringArray, data interface{}, config *MessageConfig) (*OnMessage, error) {
config, err := MergeMessageConfig(m.config.MessageConfig, config)
if err != nil {
return nil, err
}
ctx, cancel := context.WithTimeout(context.Background(), time.Duration(m.config.ResolverTimeout)*time.Millisecond)
defer cancel()
destArr, err := m.ResolveDestsContext(ctx, dests)
if err != nil {
return nil, err
}
payload, ok := data.(*payloads.Payload)
if !ok {
payload, err = newMessagePayload(data, config.MessageID, config.NoReply)
if err != nil {
return nil, err
}
}
var lock sync.Mutex
var errs error
var onRawReply *OnMessage
onReply := NewOnMessage(1, nil)
clients := m.GetClients()
if !config.NoReply {
onRawReply = NewOnMessage(1, nil)
// response channel is added first to prevent some client fail to handle response if send finish before receive response
for _, client := range clients {
client.responseChannels.Add(string(payload.MessageId), onRawReply, cache.DefaultExpiration)
}
}
success := make(chan struct{}, 1)
fail := make(chan struct{}, 1)
go func() {
sent := 0
for clientID := range clients {
err := m.sendWithClient(clientID, destArr.Elems(), payload, !config.Unencrypted, config.MaxHoldingSeconds)
if err == nil {
select {
case success <- struct{}{}:
default:
}
sent++
} else {
lock.Lock()
errs = multierror.Append(errs, err)
lock.Unlock()
}
}
if sent == 0 {
select {
case fail <- struct{}{}:
default:
}
}
if !config.NoReply {
msg := <-onRawReply.C
msg.Src, _ = removeIdentifier(msg.Src)
onReply.receive(msg, false)
}
}()
select {
case <-success:
return onReply, nil
case <-fail:
return nil, errs
}
}
// SendBinary is a wrapper of Send without interface type for gomobile
// compatibility.
func (m *MultiClient) SendBinary(dests *nkngomobile.StringArray, data []byte, config *MessageConfig) (*OnMessage, error) {
return m.Send(dests, data, config)
}
// SendText is a wrapper of Send without interface type for gomobile
// compatibility.
func (m *MultiClient) SendText(dests *nkngomobile.StringArray, data string, config *MessageConfig) (*OnMessage, error) {
return m.Send(dests, data, config)
}
// SendPayload is a wrapper of Send without interface type for gomobile
// compatibility.
func (m *MultiClient) SendPayload(dests *nkngomobile.StringArray, payload *payloads.Payload, config *MessageConfig) (*OnMessage, error) {
return m.Send(dests, payload, config)
}
func (m *MultiClient) send(dests []string, payload *payloads.Payload, encrypted bool, maxHoldingSeconds int32) error {
var lock sync.Mutex
var errs error
success := make(chan struct{}, 1)
fail := make(chan struct{}, 1)
go func() {
sent := 0
for clientID := range m.GetClients() {
err := m.sendWithClient(clientID, dests, payload, encrypted, maxHoldingSeconds)
if err == nil {
select {
case success <- struct{}{}:
default:
}
sent++
} else {
lock.Lock()
errs = multierror.Append(errs, err)
lock.Unlock()
}
}
if sent == 0 {
select {
case fail <- struct{}{}:
default:
}
}
}()
select {
case <-success:
return nil
case <-fail:
return errs
}
}
// ResolveDest wraps ResolveDestContext with background context
func (m *MultiClient) ResolveDest(dest string) (string, error) {
return m.ResolveDestContext(context.Background(), dest)
}
// ResolveDestContext resolvers an address, returns NKN address
func (m *MultiClient) ResolveDestContext(ctx context.Context, dest string) (string, error) {
return ResolveDestN(ctx, dest, m.resolvers, m.config.ResolverDepth)
}
// ResolveDests wraps ResolveDestsContext with background context
func (m *MultiClient) ResolveDests(dests *nkngomobile.StringArray) (*nkngomobile.StringArray, error) {
return m.ResolveDestsContext(context.Background(), dests)
}
// ResolveDestsContext resolvers multiple addresses
func (m *MultiClient) ResolveDestsContext(ctx context.Context, dests *nkngomobile.StringArray) (*nkngomobile.StringArray, error) {
destArr, err := ResolveDests(ctx, dests.Elems(), m.resolvers, m.config.ResolverDepth)
if err != nil {
return nil, err
}
return nkngomobile.NewStringArray(destArr...), nil
}
// Publish sends bytes or string data to all subscribers of a topic with an
// optional config.
func (m *MultiClient) Publish(topic string, data interface{}, config *MessageConfig) error {
return publish(m, topic, data, config)
}
// PublishBinary is a wrapper of Publish without interface type for gomobile
// compatibility.
func (m *MultiClient) PublishBinary(topic string, data []byte, config *MessageConfig) error {
return m.Publish(topic, data, config)
}
// PublishText is a wrapper of Publish without interface type for gomobile
// compatibility.
func (m *MultiClient) PublishText(topic string, data string, config *MessageConfig) error {
return m.Publish(topic, data, config)
}
func (m *MultiClient) newSession(remoteAddr string, sessionID []byte, config *ncp.Config) (*ncp.Session, error) {
rawClients := m.GetClients()
clientIDs := make([]string, 0, len(rawClients))
clients := make(map[string]*Client, len(rawClients))
for id, client := range rawClients {
clientID := addIdentifier("", id)
clientIDs = append(clientIDs, clientID)
clients[clientID] = client
}
sort.Strings(clientIDs)
return ncp.NewSession(m.addr, NewClientAddr(remoteAddr), clientIDs, nil, func(localClientID, remoteClientID string, buf []byte, writeTimeout time.Duration) error {
payload := &payloads.Payload{
Type: payloads.PayloadType_SESSION,
MessageId: sessionID,
Data: buf,
}
client := clients[localClientID]
err := client.sendTimeout([]string{addIdentifierPrefix(remoteAddr, remoteClientID)}, payload, true, 0, writeTimeout)
if err != nil {
return err
}
return nil
}, config)
}
func (m *MultiClient) shouldAcceptAddr(addr string) bool {
for _, allowAddr := range m.acceptAddrs {
if allowAddr.MatchString(addr) {
return true
}
}
return false
}
func (m *MultiClient) handleSessionMsg(localClientID, src string, sessionID, data []byte) error {
remoteAddr, remoteClientID := removeIdentifier(src)
sessionKey := sessionKey(remoteAddr, sessionID)
var err error
m.sessionLock.Lock()
session, ok := m.sessions[sessionKey]
if !ok {
if !m.shouldAcceptAddr(remoteAddr) {
m.sessionLock.Unlock()
return ErrAddrNotAllowed
}
session, err = m.newSession(remoteAddr, sessionID, m.config.SessionConfig)
if err != nil {
m.sessionLock.Unlock()
return err
}
m.sessions[sessionKey] = session
}
m.sessionLock.Unlock()
err = session.ReceiveWith(localClientID, remoteClientID, data)
if err != nil {
return err
}
if !ok {
select {
case m.acceptSession <- session:
default:
log.Println("Accept session channel full, discard request...")
}
}
return nil
}
// Listen will make multiclient start accepting sessions from address that
// matches any of the given regular expressions. If addrsRe is nil, any address
// will be accepted. Each function call will overwrite previous listening
// addresses.
func (m *MultiClient) Listen(addrsRe *nkngomobile.StringArray) error {
var addrs []string
if addrsRe == nil {
addrs = []string{DefaultSessionAllowAddr}
} else {
addrs = addrsRe.Elems()
}
var err error
acceptAddrs := make([]*regexp.Regexp, len(addrs))
for i := 0; i < len(acceptAddrs); i++ {
acceptAddrs[i], err = regexp.Compile(addrs[i])
if err != nil {
return err
}
}
m.lock.Lock()
m.acceptAddrs = acceptAddrs
m.lock.Unlock()
return nil
}
// Dial is the same as DialSession, but return type is net.Conn interface.
func (m *MultiClient) Dial(remoteAddr string) (net.Conn, error) {
return m.DialSession(remoteAddr)
}
// DialSession dials a session to a remote client address using this
// multiclient's dial config.
func (m *MultiClient) DialSession(remoteAddr string) (*ncp.Session, error) {
return m.DialWithConfig(remoteAddr, nil)
}
// DialWithConfig dials a session with a dial config. For any zero value field
// in config, this default dial config value of this multiclient will be used.
// If config is nil, the default dial config of this multiclient will be used.
func (m *MultiClient) DialWithConfig(remoteAddr string, config *DialConfig) (*ncp.Session, error) {
config, err := MergeDialConfig(m.config.SessionConfig, config)
if err != nil {
return nil, err
}
sessionID, err := RandomBytes(SessionIDSize)
if err != nil {
return nil, err
}
sessionKey := sessionKey(remoteAddr, sessionID)
session, err := m.newSession(remoteAddr, sessionID, config.SessionConfig)
if err != nil {
return nil, err
}
m.sessionLock.Lock()
m.sessions[sessionKey] = session
m.sessionLock.Unlock()
ctx := context.Background()
var cancel context.CancelFunc
if config.DialTimeout > 0 {
ctx, cancel = context.WithTimeout(ctx, time.Duration(config.DialTimeout)*time.Millisecond)
defer cancel()
}
err = session.Dial(ctx)
if err != nil {
return nil, err
}
return session, nil
}
// AcceptSession will wait and return the first incoming session from allowed
// remote addresses. If multiclient is closed, it will return immediately with
// ErrClosed.
func (m *MultiClient) AcceptSession() (*ncp.Session, error) {
for {
select {
case session := <-m.acceptSession:
err := session.Accept()
if err != nil {
log.Println(err)
continue
}
return session, nil
case <-m.onClose:
return nil, ErrClosed
}
}
}
// Accept is the same as AcceptSession, but the return type is net.Conn
// interface.
func (m *MultiClient) Accept() (net.Conn, error) {
return m.AcceptSession()
}
// Close closes the multiclient, including all clients it created and all
// sessions dialed and accepted. Calling close multiple times is allowed and
// will not have any effect.
func (m *MultiClient) Close() error {
m.lock.Lock()
defer m.lock.Unlock()
if m.isClosed {
return nil
}
m.sessionLock.Lock()
for _, session := range m.sessions {
err := session.Close()
if err != nil {
log.Println(err)
continue
}
}
m.sessionLock.Unlock()
time.AfterFunc(time.Duration(m.config.SessionConfig.Linger)*time.Millisecond, func() {
for _, client := range m.GetClients() {
err := client.Close()
if err != nil {
log.Println(err)
continue
}
}
})
m.isClosed = true
m.OnConnect.close()
m.OnMessage.close()
close(m.onClose)
return nil
}
// IsClosed returns whether this multiclient is closed.
func (m *MultiClient) IsClosed() bool {
m.lock.RLock()
defer m.lock.RUnlock()
return m.isClosed
}
// Reconnect forces all clients to find node and connect again.
func (m *MultiClient) Reconnect() {
m.lock.RLock()
defer m.lock.RUnlock()
if m.isClosed {
return
}
for _, client := range m.clients {
client.Reconnect(nil)
}
}
func (m *MultiClient) getConfig() *ClientConfig {
return m.config
}
// SignTransaction signs an unsigned transaction using this multiclient's key
// pair.
func (m *MultiClient) SignTransaction(tx *transaction.Transaction) error {
return m.GetDefaultClient().SignTransaction(tx)
}
// NewNanoPay is a shortcut for NewNanoPay using this multiclient's wallet
// address as sender.
//
// Duration is changed to signed int for gomobile compatibility.
func (m *MultiClient) NewNanoPay(recipientAddress, fee string, duration int) (*NanoPay, error) {
return NewNanoPay(m, m.GetDefaultClient().wallet, recipientAddress, fee, duration)
}
// NewNanoPayClaimer is a shortcut for NewNanoPayClaimer using this multiclient
// as RPC client.
func (m *MultiClient) NewNanoPayClaimer(recipientAddress string, claimIntervalMs, lingerMs int32, minFlushAmount string, onError *OnError) (*NanoPayClaimer, error) {
if len(recipientAddress) == 0 {
recipientAddress = m.GetDefaultClient().wallet.Address()
}
return NewNanoPayClaimer(m, recipientAddress, claimIntervalMs, lingerMs, minFlushAmount, onError)
}
// GetNonce wraps GetNonceContext with background context.
func (m *MultiClient) GetNonce(txPool bool) (int64, error) {
return m.GetNonceContext(context.Background(), txPool)
}
// GetNonceContext is the same as package level GetNonceContext, but using
// connected node as the RPC server, followed by this multiclient's
// SeedRPCServerAddr if failed.
func (m *MultiClient) GetNonceContext(ctx context.Context, txPool bool) (int64, error) {
return m.GetNonceByAddressContext(ctx, m.GetDefaultClient().wallet.Address(), txPool)
}
// GetNonceByAddress wraps GetNonceByAddressContext with background context.
func (m *MultiClient) GetNonceByAddress(address string, txPool bool) (int64, error) {
return m.GetNonceByAddressContext(context.Background(), address, txPool)
}
// GetNonceByAddressContext is the same as package level GetNonceContext, but
// using connected node as the RPC server, followed by this multiclient's
// SeedRPCServerAddr if failed.
func (m *MultiClient) GetNonceByAddressContext(ctx context.Context, address string, txPool bool) (int64, error) {
for _, c := range m.GetClients() {
if c.wallet.config.SeedRPCServerAddr.Len() > 0 {
res, err := GetNonceContext(ctx, address, txPool, c.wallet.config)
if err == nil {
return res, err
}
}
}
return GetNonceContext(ctx, address, txPool, m.config)
}
// GetHeight wraps GetHeightContext with background context.
func (m *MultiClient) GetHeight() (int32, error) {
return m.GetHeightContext(context.Background())
}
// GetHeightContext is the same as package level GetHeightContext, but using
// connected node as the RPC server, followed by this multiclient's
// SeedRPCServerAddr if failed.
func (m *MultiClient) GetHeightContext(ctx context.Context) (int32, error) {
for _, c := range m.GetClients() {
if c.wallet.config.SeedRPCServerAddr.Len() > 0 {
res, err := GetHeightContext(ctx, c.wallet.config)
if err == nil {
return res, err
}
}
}
return GetHeightContext(ctx, m.config)
}
// Balance wraps BalanceContext with background context.
func (m *MultiClient) Balance() (*Amount, error) {
return m.BalanceContext(context.Background())
}
// BalanceContext is the same as package level GetBalanceContext, but using
// connected node as the RPC server, followed by this multiclient's
// SeedRPCServerAddr if failed.
func (m *MultiClient) BalanceContext(ctx context.Context) (*Amount, error) {
return m.BalanceByAddressContext(ctx, m.GetDefaultClient().wallet.Address())
}
// BalanceByAddress wraps BalanceByAddressContext with background context.
func (m *MultiClient) BalanceByAddress(address string) (*Amount, error) {
return m.BalanceByAddressContext(context.Background(), address)
}
// BalanceByAddressContext is the same as package level GetBalanceContext, but
// using connected node as the RPC server, followed by this multiclient's
// SeedRPCServerAddr if failed.
func (m *MultiClient) BalanceByAddressContext(ctx context.Context, address string) (*Amount, error) {
for _, c := range m.GetClients() {
if c.wallet.config.SeedRPCServerAddr.Len() > 0 {
res, err := GetBalanceContext(ctx, address, c.wallet.config)
if err == nil {
return res, err
}
}
}
return GetBalanceContext(ctx, address, m.config)
}
// GetSubscribers wraps GetSubscribersContext with background context.
func (m *MultiClient) GetSubscribers(topic string, offset, limit int, meta, txPool bool, subscriberHashPrefix []byte) (*Subscribers, error) {
return m.GetSubscribersContext(context.Background(), topic, offset, limit, meta, txPool, subscriberHashPrefix)
}
// GetSubscribersContext is the same as package level GetSubscribersContext, but
// using connected node as the RPC server, followed by this multiclient's
// SeedRPCServerAddr if failed.
func (m *MultiClient) GetSubscribersContext(ctx context.Context, topic string, offset, limit int, meta, txPool bool, subscriberHashPrefix []byte) (*Subscribers, error) {
for _, c := range m.GetClients() {
if c.wallet.config.SeedRPCServerAddr.Len() > 0 {
res, err := GetSubscribersContext(ctx, topic, offset, limit, meta, txPool, subscriberHashPrefix, c.wallet.config)
if err == nil {
return res, err
}
}
}
return GetSubscribersContext(ctx, topic, offset, limit, meta, txPool, subscriberHashPrefix, m.config)
}
// GetSubscription wraps GetSubscriptionContext with background context.
func (m *MultiClient) GetSubscription(topic string, subscriber string) (*Subscription, error) {
return m.GetSubscriptionContext(context.Background(), topic, subscriber)
}
// GetSubscriptionContext is the same as package level GetSubscriptionContext,
// but using connected node as the RPC server, followed by this multiclient's
// SeedRPCServerAddr if failed.
func (m *MultiClient) GetSubscriptionContext(ctx context.Context, topic string, subscriber string) (*Subscription, error) {
for _, c := range m.GetClients() {
if c.wallet.config.SeedRPCServerAddr.Len() > 0 {
res, err := GetSubscriptionContext(ctx, topic, subscriber, c.wallet.config)
if err == nil {
return res, err
}
}
}
return GetSubscriptionContext(ctx, topic, subscriber, m.config)
}
// GetSubscribersCount wraps GetSubscribersCountContext with background context.
func (m *MultiClient) GetSubscribersCount(topic string, subscriberHashPrefix []byte) (int, error) {
return m.GetSubscribersCountContext(context.Background(), topic, subscriberHashPrefix)
}
// GetSubscribersCountContext is the same as package level
// GetSubscribersCountContext, but using connected node as the RPC server,