forked from vcabbage/amqp
-
Notifications
You must be signed in to change notification settings - Fork 2
/
types.go
4188 lines (3592 loc) · 122 KB
/
types.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 amqp
import (
"encoding/binary"
"encoding/hex"
"fmt"
"math"
"reflect"
"strconv"
"time"
"unicode/utf8"
)
type amqpType uint8
// Type codes
const (
typeCodeNull amqpType = 0x40
// Bool
typeCodeBool amqpType = 0x56 // boolean with the octet 0x00 being false and octet 0x01 being true
typeCodeBoolTrue amqpType = 0x41
typeCodeBoolFalse amqpType = 0x42
// Unsigned
typeCodeUbyte amqpType = 0x50 // 8-bit unsigned integer (1)
typeCodeUshort amqpType = 0x60 // 16-bit unsigned integer in network byte order (2)
typeCodeUint amqpType = 0x70 // 32-bit unsigned integer in network byte order (4)
typeCodeSmallUint amqpType = 0x52 // unsigned integer value in the range 0 to 255 inclusive (1)
typeCodeUint0 amqpType = 0x43 // the uint value 0 (0)
typeCodeUlong amqpType = 0x80 // 64-bit unsigned integer in network byte order (8)
typeCodeSmallUlong amqpType = 0x53 // unsigned long value in the range 0 to 255 inclusive (1)
typeCodeUlong0 amqpType = 0x44 // the ulong value 0 (0)
// Signed
typeCodeByte amqpType = 0x51 // 8-bit two's-complement integer (1)
typeCodeShort amqpType = 0x61 // 16-bit two's-complement integer in network byte order (2)
typeCodeInt amqpType = 0x71 // 32-bit two's-complement integer in network byte order (4)
typeCodeSmallint amqpType = 0x54 // 8-bit two's-complement integer (1)
typeCodeLong amqpType = 0x81 // 64-bit two's-complement integer in network byte order (8)
typeCodeSmalllong amqpType = 0x55 // 8-bit two's-complement integer
// Decimal
typeCodeFloat amqpType = 0x72 // IEEE 754-2008 binary32 (4)
typeCodeDouble amqpType = 0x82 // IEEE 754-2008 binary64 (8)
typeCodeDecimal32 amqpType = 0x74 // IEEE 754-2008 decimal32 using the Binary Integer Decimal encoding (4)
typeCodeDecimal64 amqpType = 0x84 // IEEE 754-2008 decimal64 using the Binary Integer Decimal encoding (8)
typeCodeDecimal128 amqpType = 0x94 // IEEE 754-2008 decimal128 using the Binary Integer Decimal encoding (16)
// Other
typeCodeChar amqpType = 0x73 // a UTF-32BE encoded Unicode character (4)
typeCodeTimestamp amqpType = 0x83 // 64-bit two's-complement integer representing milliseconds since the unix epoch
typeCodeUUID amqpType = 0x98 // UUID as defined in section 4.1.2 of RFC-4122
// Variable Length
typeCodeVbin8 amqpType = 0xa0 // up to 2^8 - 1 octets of binary data (1 + variable)
typeCodeVbin32 amqpType = 0xb0 // up to 2^32 - 1 octets of binary data (4 + variable)
typeCodeStr8 amqpType = 0xa1 // up to 2^8 - 1 octets worth of UTF-8 Unicode (with no byte order mark) (1 + variable)
typeCodeStr32 amqpType = 0xb1 // up to 2^32 - 1 octets worth of UTF-8 Unicode (with no byte order mark) (4 +variable)
typeCodeSym8 amqpType = 0xa3 // up to 2^8 - 1 seven bit ASCII characters representing a symbolic value (1 + variable)
typeCodeSym32 amqpType = 0xb3 // up to 2^32 - 1 seven bit ASCII characters representing a symbolic value (4 + variable)
// Compound
typeCodeList0 amqpType = 0x45 // the empty list (i.e. the list with no elements) (0)
typeCodeList8 amqpType = 0xc0 // up to 2^8 - 1 list elements with total size less than 2^8 octets (1 + compound)
typeCodeList32 amqpType = 0xd0 // up to 2^32 - 1 list elements with total size less than 2^32 octets (4 + compound)
typeCodeMap8 amqpType = 0xc1 // up to 2^8 - 1 octets of encoded map data (1 + compound)
typeCodeMap32 amqpType = 0xd1 // up to 2^32 - 1 octets of encoded map data (4 + compound)
typeCodeArray8 amqpType = 0xe0 // up to 2^8 - 1 array elements with total size less than 2^8 octets (1 + array)
typeCodeArray32 amqpType = 0xf0 // up to 2^32 - 1 array elements with total size less than 2^32 octets (4 + array)
// Composites
typeCodeOpen amqpType = 0x10
typeCodeBegin amqpType = 0x11
typeCodeAttach amqpType = 0x12
typeCodeFlow amqpType = 0x13
typeCodeTransfer amqpType = 0x14
typeCodeDisposition amqpType = 0x15
typeCodeDetach amqpType = 0x16
typeCodeEnd amqpType = 0x17
typeCodeClose amqpType = 0x18
typeCodeSource amqpType = 0x28
typeCodeTarget amqpType = 0x29
typeCodeError amqpType = 0x1d
typeCodeMessageHeader amqpType = 0x70
typeCodeDeliveryAnnotations amqpType = 0x71
typeCodeMessageAnnotations amqpType = 0x72
typeCodeMessageProperties amqpType = 0x73
typeCodeApplicationProperties amqpType = 0x74
typeCodeApplicationData amqpType = 0x75
typeCodeAMQPSequence amqpType = 0x76
typeCodeAMQPValue amqpType = 0x77
typeCodeFooter amqpType = 0x78
typeCodeStateReceived amqpType = 0x23
typeCodeStateAccepted amqpType = 0x24
typeCodeStateRejected amqpType = 0x25
typeCodeStateReleased amqpType = 0x26
typeCodeStateModified amqpType = 0x27
typeCodeSASLMechanism amqpType = 0x40
typeCodeSASLInit amqpType = 0x41
typeCodeSASLChallenge amqpType = 0x42
typeCodeSASLResponse amqpType = 0x43
typeCodeSASLOutcome amqpType = 0x44
typeCodeDeleteOnClose amqpType = 0x2b
typeCodeDeleteOnNoLinks amqpType = 0x2c
typeCodeDeleteOnNoMessages amqpType = 0x2d
typeCodeDeleteOnNoLinksOrMessages amqpType = 0x2e
)
// Frame structure:
//
// header (8 bytes)
// 0-3: SIZE (total size, at least 8 bytes for header, uint32)
// 4: DOFF (data offset,at least 2, count of 4 bytes words, uint8)
// 5: TYPE (frame type)
// 0x0: AMQP
// 0x1: SASL
// 6-7: type dependent (channel for AMQP)
// extended header (opt)
// body (opt)
// frameHeader in a structure appropriate for use with binary.Read()
type frameHeader struct {
// size: an unsigned 32-bit integer that MUST contain the total frame size of the frame header,
// extended header, and frame body. The frame is malformed if the size is less than the size of
// the frame header (8 bytes).
Size uint32
// doff: gives the position of the body within the frame. The value of the data offset is an
// unsigned, 8-bit integer specifying a count of 4-byte words. Due to the mandatory 8-byte
// frame header, the frame is malformed if the value is less than 2.
DataOffset uint8
FrameType uint8
Channel uint16
}
const (
frameTypeAMQP = 0x0
frameTypeSASL = 0x1
frameHeaderSize = 8
)
type protoHeader struct {
ProtoID protoID
Major uint8
Minor uint8
Revision uint8
}
// frame is the decoded representation of a frame
type frame struct {
type_ uint8 // AMQP/SASL
channel uint16 // channel this frame is for
body frameBody // body of the frame
// optional channel which will be closed after net transmit
done chan deliveryState
}
// frameBody adds some type safety to frame encoding
type frameBody interface {
frameBody()
}
/*
<type name="open" class="composite" source="list" provides="frame">
<descriptor name="amqp:open:list" code="0x00000000:0x00000010"/>
<field name="container-id" type="string" mandatory="true"/>
<field name="hostname" type="string"/>
<field name="max-frame-size" type="uint" default="4294967295"/>
<field name="channel-max" type="ushort" default="65535"/>
<field name="idle-time-out" type="milliseconds"/>
<field name="outgoing-locales" type="ietf-language-tag" multiple="true"/>
<field name="incoming-locales" type="ietf-language-tag" multiple="true"/>
<field name="offered-capabilities" type="symbol" multiple="true"/>
<field name="desired-capabilities" type="symbol" multiple="true"/>
<field name="properties" type="fields"/>
</type>
*/
type performOpen struct {
ContainerID string // required
Hostname string
MaxFrameSize uint32 // default: 4294967295
ChannelMax uint16 // default: 65535
IdleTimeout time.Duration // from milliseconds
OutgoingLocales multiSymbol
IncomingLocales multiSymbol
OfferedCapabilities multiSymbol
DesiredCapabilities multiSymbol
Properties map[symbol]interface{}
}
func (o *performOpen) frameBody() {}
func (o *performOpen) marshal(wr *buffer) error {
return marshalComposite(wr, typeCodeOpen, []marshalField{
{value: &o.ContainerID, omit: false},
{value: &o.Hostname, omit: o.Hostname == ""},
{value: &o.MaxFrameSize, omit: o.MaxFrameSize == 4294967295},
{value: &o.ChannelMax, omit: o.ChannelMax == 65535},
{value: (*milliseconds)(&o.IdleTimeout), omit: o.IdleTimeout == 0},
{value: &o.OutgoingLocales, omit: len(o.OutgoingLocales) == 0},
{value: &o.IncomingLocales, omit: len(o.IncomingLocales) == 0},
{value: &o.OfferedCapabilities, omit: len(o.OfferedCapabilities) == 0},
{value: &o.DesiredCapabilities, omit: len(o.DesiredCapabilities) == 0},
{value: o.Properties, omit: len(o.Properties) == 0},
})
}
func (o *performOpen) unmarshal(r *buffer) error {
return unmarshalComposite(r, typeCodeOpen, []unmarshalField{
{field: &o.ContainerID, handleNull: func() error { return errorNew("Open.ContainerID is required") }},
{field: &o.Hostname},
{field: &o.MaxFrameSize, handleNull: func() error { o.MaxFrameSize = 4294967295; return nil }},
{field: &o.ChannelMax, handleNull: func() error { o.ChannelMax = 65535; return nil }},
{field: (*milliseconds)(&o.IdleTimeout)},
{field: &o.OutgoingLocales},
{field: &o.IncomingLocales},
{field: &o.OfferedCapabilities},
{field: &o.DesiredCapabilities},
{field: &o.Properties},
}...)
}
func (o *performOpen) String() string {
return fmt.Sprintf("Open{ContainerID : %s, Hostname: %s, MaxFrameSize: %d, "+
"ChannelMax: %d, IdleTimeout: %v, "+
"OutgoingLocales: %v, IncomingLocales: %v, "+
"OfferedCapabilities: %v, DesiredCapabilities: %v, "+
"Properties: %v}",
o.ContainerID,
o.Hostname,
o.MaxFrameSize,
o.ChannelMax,
o.IdleTimeout,
o.OutgoingLocales,
o.IncomingLocales,
o.OfferedCapabilities,
o.DesiredCapabilities,
o.Properties,
)
}
/*
<type name="begin" class="composite" source="list" provides="frame">
<descriptor name="amqp:begin:list" code="0x00000000:0x00000011"/>
<field name="remote-channel" type="ushort"/>
<field name="next-outgoing-id" type="transfer-number" mandatory="true"/>
<field name="incoming-window" type="uint" mandatory="true"/>
<field name="outgoing-window" type="uint" mandatory="true"/>
<field name="handle-max" type="handle" default="4294967295"/>
<field name="offered-capabilities" type="symbol" multiple="true"/>
<field name="desired-capabilities" type="symbol" multiple="true"/>
<field name="properties" type="fields"/>
</type>
*/
type performBegin struct {
// the remote channel for this session
// If a session is locally initiated, the remote-channel MUST NOT be set.
// When an endpoint responds to a remotely initiated session, the remote-channel
// MUST be set to the channel on which the remote session sent the begin.
RemoteChannel *uint16
// the transfer-id of the first transfer id the sender will send
NextOutgoingID uint32 // required, sequence number http://www.ietf.org/rfc/rfc1982.txt
// the initial incoming-window of the sender
IncomingWindow uint32 // required
// the initial outgoing-window of the sender
OutgoingWindow uint32 // required
// the maximum handle value that can be used on the session
// The handle-max value is the highest handle value that can be
// used on the session. A peer MUST NOT attempt to attach a link
// using a handle value outside the range that its partner can handle.
// A peer that receives a handle outside the supported range MUST
// close the connection with the framing-error error-code.
HandleMax uint32 // default 4294967295
// the extension capabilities the sender supports
// http://www.amqp.org/specification/1.0/session-capabilities
OfferedCapabilities multiSymbol
// the extension capabilities the sender can use if the receiver supports them
// The sender MUST NOT attempt to use any capability other than those it
// has declared in desired-capabilities field.
DesiredCapabilities multiSymbol
// session properties
// http://www.amqp.org/specification/1.0/session-properties
Properties map[symbol]interface{}
}
func (b *performBegin) frameBody() {}
func (b *performBegin) String() string {
return fmt.Sprintf("Begin{RemoteChannel: %v, NextOutgoingID: %d, IncomingWindow: %d, "+
"OutgoingWindow: %d, HandleMax: %d, OfferedCapabilities: %v, DesiredCapabilities: %v, "+
"Properties: %v}",
formatUint16Ptr(b.RemoteChannel),
b.NextOutgoingID,
b.IncomingWindow,
b.OutgoingWindow,
b.HandleMax,
b.OfferedCapabilities,
b.DesiredCapabilities,
b.Properties,
)
}
func formatUint16Ptr(p *uint16) string {
if p == nil {
return "<nil>"
}
return strconv.FormatUint(uint64(*p), 10)
}
func (b *performBegin) marshal(wr *buffer) error {
return marshalComposite(wr, typeCodeBegin, []marshalField{
{value: b.RemoteChannel, omit: b.RemoteChannel == nil},
{value: &b.NextOutgoingID, omit: false},
{value: &b.IncomingWindow, omit: false},
{value: &b.OutgoingWindow, omit: false},
{value: &b.HandleMax, omit: b.HandleMax == 4294967295},
{value: &b.OfferedCapabilities, omit: len(b.OfferedCapabilities) == 0},
{value: &b.DesiredCapabilities, omit: len(b.DesiredCapabilities) == 0},
{value: b.Properties, omit: b.Properties == nil},
})
}
func (b *performBegin) unmarshal(r *buffer) error {
return unmarshalComposite(r, typeCodeBegin, []unmarshalField{
{field: &b.RemoteChannel},
{field: &b.NextOutgoingID, handleNull: func() error { return errorNew("Begin.NextOutgoingID is required") }},
{field: &b.IncomingWindow, handleNull: func() error { return errorNew("Begin.IncomingWindow is required") }},
{field: &b.OutgoingWindow, handleNull: func() error { return errorNew("Begin.OutgoingWindow is required") }},
{field: &b.HandleMax, handleNull: func() error { b.HandleMax = 4294967295; return nil }},
{field: &b.OfferedCapabilities},
{field: &b.DesiredCapabilities},
{field: &b.Properties},
}...)
}
/*
<type name="attach" class="composite" source="list" provides="frame">
<descriptor name="amqp:attach:list" code="0x00000000:0x00000012"/>
<field name="name" type="string" mandatory="true"/>
<field name="handle" type="handle" mandatory="true"/>
<field name="role" type="role" mandatory="true"/>
<field name="snd-settle-mode" type="sender-settle-mode" default="mixed"/>
<field name="rcv-settle-mode" type="receiver-settle-mode" default="first"/>
<field name="source" type="*" requires="source"/>
<field name="target" type="*" requires="target"/>
<field name="unsettled" type="map"/>
<field name="incomplete-unsettled" type="boolean" default="false"/>
<field name="initial-delivery-count" type="sequence-no"/>
<field name="max-message-size" type="ulong"/>
<field name="offered-capabilities" type="symbol" multiple="true"/>
<field name="desired-capabilities" type="symbol" multiple="true"/>
<field name="properties" type="fields"/>
</type>
*/
type performAttach struct {
// the name of the link
//
// This name uniquely identifies the link from the container of the source
// to the container of the target node, e.g., if the container of the source
// node is A, and the container of the target node is B, the link MAY be
// globally identified by the (ordered) tuple (A,B,<name>).
Name string // required
// the handle for the link while attached
//
// The numeric handle assigned by the the peer as a shorthand to refer to the
// link in all performatives that reference the link until the it is detached.
//
// The handle MUST NOT be used for other open links. An attempt to attach using
// a handle which is already associated with a link MUST be responded to with
// an immediate close carrying a handle-in-use session-error.
//
// To make it easier to monitor AMQP link attach frames, it is RECOMMENDED that
// implementations always assign the lowest available handle to this field.
//
// The two endpoints MAY potentially use different handles to refer to the same link.
// Link handles MAY be reused once a link is closed for both send and receive.
Handle uint32 // required
// role of the link endpoint
//
// The role being played by the peer, i.e., whether the peer is the sender or the
// receiver of messages on the link.
Role role
// settlement policy for the sender
//
// The delivery settlement policy for the sender. When set at the receiver this
// indicates the desired value for the settlement mode at the sender. When set
// at the sender this indicates the actual settlement mode in use. The sender
// SHOULD respect the receiver's desired settlement mode if the receiver initiates
// the attach exchange and the sender supports the desired mode.
//
// 0: unsettled - The sender will send all deliveries initially unsettled to the receiver.
// 1: settled - The sender will send all deliveries settled to the receiver.
// 2: mixed - The sender MAY send a mixture of settled and unsettled deliveries to the receiver.
SenderSettleMode *SenderSettleMode
// the settlement policy of the receiver
//
// The delivery settlement policy for the receiver. When set at the sender this
// indicates the desired value for the settlement mode at the receiver.
// When set at the receiver this indicates the actual settlement mode in use.
// The receiver SHOULD respect the sender's desired settlement mode if the sender
// initiates the attach exchange and the receiver supports the desired mode.
//
// 0: first - The receiver will spontaneously settle all incoming transfers.
// 1: second - The receiver will only settle after sending the disposition to
// the sender and receiving a disposition indicating settlement of
// the delivery from the sender.
ReceiverSettleMode *ReceiverSettleMode
// the source for messages
//
// If no source is specified on an outgoing link, then there is no source currently
// attached to the link. A link with no source will never produce outgoing messages.
Source *source
// the target for messages
//
// If no target is specified on an incoming link, then there is no target currently
// attached to the link. A link with no target will never permit incoming messages.
Target *target
// unsettled delivery state
//
// This is used to indicate any unsettled delivery states when a suspended link is
// resumed. The map is keyed by delivery-tag with values indicating the delivery state.
// The local and remote delivery states for a given delivery-tag MUST be compared to
// resolve any in-doubt deliveries. If necessary, deliveries MAY be resent, or resumed
// based on the outcome of this comparison. See subsection 2.6.13.
//
// If the local unsettled map is too large to be encoded within a frame of the agreed
// maximum frame size then the session MAY be ended with the frame-size-too-small error.
// The endpoint SHOULD make use of the ability to send an incomplete unsettled map
// (see below) to avoid sending an error.
//
// The unsettled map MUST NOT contain null valued keys.
//
// When reattaching (as opposed to resuming), the unsettled map MUST be null.
Unsettled unsettled
// If set to true this field indicates that the unsettled map provided is not complete.
// When the map is incomplete the recipient of the map cannot take the absence of a
// delivery tag from the map as evidence of settlement. On receipt of an incomplete
// unsettled map a sending endpoint MUST NOT send any new deliveries (i.e. deliveries
// where resume is not set to true) to its partner (and a receiving endpoint which sent
// an incomplete unsettled map MUST detach with an error on receiving a transfer which
// does not have the resume flag set to true).
//
// Note that if this flag is set to true then the endpoints MUST detach and reattach at
// least once in order to send new deliveries. This flag can be useful when there are
// too many entries in the unsettled map to fit within a single frame. An endpoint can
// attach, resume, settle, and detach until enough unsettled state has been cleared for
// an attach where this flag is set to false.
IncompleteUnsettled bool // default: false
// the sender's initial value for delivery-count
//
// This MUST NOT be null if role is sender, and it is ignored if the role is receiver.
InitialDeliveryCount uint32 // sequence number
// the maximum message size supported by the link endpoint
//
// This field indicates the maximum message size supported by the link endpoint.
// Any attempt to deliver a message larger than this results in a message-size-exceeded
// link-error. If this field is zero or unset, there is no maximum size imposed by the
// link endpoint.
MaxMessageSize uint64
// the extension capabilities the sender supports
// http://www.amqp.org/specification/1.0/link-capabilities
OfferedCapabilities multiSymbol
// the extension capabilities the sender can use if the receiver supports them
//
// The sender MUST NOT attempt to use any capability other than those it
// has declared in desired-capabilities field.
DesiredCapabilities multiSymbol
// link properties
// http://www.amqp.org/specification/1.0/link-properties
Properties map[symbol]interface{}
}
func (a *performAttach) frameBody() {}
func (a performAttach) String() string {
return fmt.Sprintf("Attach{Name: %s, Handle: %d, Role: %s, SenderSettleMode: %s, ReceiverSettleMode: %s, "+
"Source: %v, Target: %v, Unsettled: %v, IncompleteUnsettled: %t, InitialDeliveryCount: %d, MaxMessageSize: %d, "+
"OfferedCapabilities: %v, DesiredCapabilities: %v, Properties: %v}",
a.Name,
a.Handle,
a.Role,
a.SenderSettleMode,
a.ReceiverSettleMode,
a.Source,
a.Target,
a.Unsettled,
a.IncompleteUnsettled,
a.InitialDeliveryCount,
a.MaxMessageSize,
a.OfferedCapabilities,
a.DesiredCapabilities,
a.Properties,
)
}
func (a *performAttach) marshal(wr *buffer) error {
return marshalComposite(wr, typeCodeAttach, []marshalField{
{value: &a.Name, omit: false},
{value: &a.Handle, omit: false},
{value: &a.Role, omit: false},
{value: a.SenderSettleMode, omit: a.SenderSettleMode == nil},
{value: a.ReceiverSettleMode, omit: a.ReceiverSettleMode == nil},
{value: a.Source, omit: a.Source == nil},
{value: a.Target, omit: a.Target == nil},
{value: a.Unsettled, omit: len(a.Unsettled) == 0},
{value: &a.IncompleteUnsettled, omit: !a.IncompleteUnsettled},
{value: &a.InitialDeliveryCount, omit: a.Role == roleReceiver},
{value: &a.MaxMessageSize, omit: a.MaxMessageSize == 0},
{value: &a.OfferedCapabilities, omit: len(a.OfferedCapabilities) == 0},
{value: &a.DesiredCapabilities, omit: len(a.DesiredCapabilities) == 0},
{value: a.Properties, omit: len(a.Properties) == 0},
})
}
func (a *performAttach) unmarshal(r *buffer) error {
return unmarshalComposite(r, typeCodeAttach, []unmarshalField{
{field: &a.Name, handleNull: func() error { return errorNew("Attach.Name is required") }},
{field: &a.Handle, handleNull: func() error { return errorNew("Attach.Handle is required") }},
{field: &a.Role, handleNull: func() error { return errorNew("Attach.Role is required") }},
{field: &a.SenderSettleMode},
{field: &a.ReceiverSettleMode},
{field: &a.Source},
{field: &a.Target},
{field: &a.Unsettled},
{field: &a.IncompleteUnsettled},
{field: &a.InitialDeliveryCount},
{field: &a.MaxMessageSize},
{field: &a.OfferedCapabilities},
{field: &a.DesiredCapabilities},
{field: &a.Properties},
}...)
}
type role bool
const (
roleSender role = false
roleReceiver role = true
)
func (rl role) String() string {
if rl {
return "Receiver"
}
return "Sender"
}
func (rl *role) unmarshal(r *buffer) error {
b, err := readBool(r)
*rl = role(b)
return err
}
func (rl role) marshal(wr *buffer) error {
return marshal(wr, (bool)(rl))
}
type deliveryState interface{} // TODO: http://docs.oasis-open.org/amqp/core/v1.0/os/amqp-core-transactions-v1.0-os.html#type-declared
type unsettled map[string]deliveryState
func (u unsettled) marshal(wr *buffer) error {
return writeMap(wr, u)
}
func (u *unsettled) unmarshal(r *buffer) error {
count, err := readMapHeader(r)
if err != nil {
return err
}
m := make(unsettled, count/2)
for i := uint32(0); i < count; i += 2 {
key, err := readString(r)
if err != nil {
return err
}
var value deliveryState
err = unmarshal(r, &value)
if err != nil {
return err
}
m[key] = value
}
*u = m
return nil
}
type filter map[symbol]*describedType
func (f filter) marshal(wr *buffer) error {
return writeMap(wr, f)
}
func (f *filter) unmarshal(r *buffer) error {
count, err := readMapHeader(r)
if err != nil {
return err
}
m := make(filter, count/2)
for i := uint32(0); i < count; i += 2 {
key, err := readString(r)
if err != nil {
return err
}
var value describedType
err = unmarshal(r, &value)
if err != nil {
return err
}
m[symbol(key)] = &value
}
*f = m
return nil
}
/*
<type name="source" class="composite" source="list" provides="source">
<descriptor name="amqp:source:list" code="0x00000000:0x00000028"/>
<field name="address" type="*" requires="address"/>
<field name="durable" type="terminus-durability" default="none"/>
<field name="expiry-policy" type="terminus-expiry-policy" default="session-end"/>
<field name="timeout" type="seconds" default="0"/>
<field name="dynamic" type="boolean" default="false"/>
<field name="dynamic-node-properties" type="node-properties"/>
<field name="distribution-mode" type="symbol" requires="distribution-mode"/>
<field name="filter" type="filter-set"/>
<field name="default-outcome" type="*" requires="outcome"/>
<field name="outcomes" type="symbol" multiple="true"/>
<field name="capabilities" type="symbol" multiple="true"/>
</type>
*/
type source struct {
// the address of the source
//
// The address of the source MUST NOT be set when sent on a attach frame sent by
// the receiving link endpoint where the dynamic flag is set to true (that is where
// the receiver is requesting the sender to create an addressable node).
//
// The address of the source MUST be set when sent on a attach frame sent by the
// sending link endpoint where the dynamic flag is set to true (that is where the
// sender has created an addressable node at the request of the receiver and is now
// communicating the address of that created node). The generated name of the address
// SHOULD include the link name and the container-id of the remote container to allow
// for ease of identification.
Address string
// indicates the durability of the terminus
//
// Indicates what state of the terminus will be retained durably: the state of durable
// messages, only existence and configuration of the terminus, or no state at all.
//
// 0: none
// 1: configuration
// 2: unsettled-state
Durable Durability
// the expiry policy of the source
//
// link-detach: The expiry timer starts when terminus is detached.
// session-end: The expiry timer starts when the most recently associated session is
// ended.
// connection-close: The expiry timer starts when most recently associated connection
// is closed.
// never: The terminus never expires.
ExpiryPolicy ExpiryPolicy
// duration that an expiring source will be retained
//
// The source starts expiring as indicated by the expiry-policy.
Timeout uint32 // seconds
// request dynamic creation of a remote node
//
// When set to true by the receiving link endpoint, this field constitutes a request
// for the sending peer to dynamically create a node at the source. In this case the
// address field MUST NOT be set.
//
// When set to true by the sending link endpoint this field indicates creation of a
// dynamically created node. In this case the address field will contain the address
// of the created node. The generated address SHOULD include the link name and other
// available information on the initiator of the request (such as the remote
// container-id) in some recognizable form for ease of traceability.
Dynamic bool
// properties of the dynamically created node
//
// If the dynamic field is not set to true this field MUST be left unset.
//
// When set by the receiving link endpoint, this field contains the desired
// properties of the node the receiver wishes to be created. When set by the
// sending link endpoint this field contains the actual properties of the
// dynamically created node. See subsection 3.5.9 for standard node properties.
// http://www.amqp.org/specification/1.0/node-properties
//
// lifetime-policy: The lifetime of a dynamically generated node.
// Definitionally, the lifetime will never be less than the lifetime
// of the link which caused its creation, however it is possible to
// extend the lifetime of dynamically created node using a lifetime
// policy. The value of this entry MUST be of a type which provides
// the lifetime-policy archetype. The following standard
// lifetime-policies are defined below: delete-on-close,
// delete-on-no-links, delete-on-no-messages or
// delete-on-no-links-or-messages.
// supported-dist-modes: The distribution modes that the node supports.
// The value of this entry MUST be one or more symbols which are valid
// distribution-modes. That is, the value MUST be of the same type as
// would be valid in a field defined with the following attributes:
// type="symbol" multiple="true" requires="distribution-mode"
DynamicNodeProperties map[symbol]interface{} // TODO: implement custom type with validation
// the distribution mode of the link
//
// This field MUST be set by the sending end of the link if the endpoint supports more
// than one distribution-mode. This field MAY be set by the receiving end of the link
// to indicate a preference when a node supports multiple distribution modes.
DistributionMode symbol
// a set of predicates to filter the messages admitted onto the link
//
// The receiving endpoint sets its desired filter, the sending endpoint sets the filter
// actually in place (including any filters defaulted at the node). The receiving
// endpoint MUST check that the filter in place meets its needs and take responsibility
// for detaching if it does not.
Filter filter
// default outcome for unsettled transfers
//
// Indicates the outcome to be used for transfers that have not reached a terminal
// state at the receiver when the transfer is settled, including when the source
// is destroyed. The value MUST be a valid outcome (e.g., released or rejected).
DefaultOutcome interface{}
// descriptors for the outcomes that can be chosen on this link
//
// The values in this field are the symbolic descriptors of the outcomes that can
// be chosen on this link. This field MAY be empty, indicating that the default-outcome
// will be assumed for all message transfers (if the default-outcome is not set, and no
// outcomes are provided, then the accepted outcome MUST be supported by the source).
//
// When present, the values MUST be a symbolic descriptor of a valid outcome,
// e.g., "amqp:accepted:list".
Outcomes multiSymbol
// the extension capabilities the sender supports/desires
//
// http://www.amqp.org/specification/1.0/source-capabilities
Capabilities multiSymbol
}
func (s *source) marshal(wr *buffer) error {
return marshalComposite(wr, typeCodeSource, []marshalField{
{value: &s.Address, omit: s.Address == ""},
{value: &s.Durable, omit: s.Durable == DurabilityNone},
{value: &s.ExpiryPolicy, omit: s.ExpiryPolicy == "" || s.ExpiryPolicy == ExpirySessionEnd},
{value: &s.Timeout, omit: s.Timeout == 0},
{value: &s.Dynamic, omit: !s.Dynamic},
{value: s.DynamicNodeProperties, omit: len(s.DynamicNodeProperties) == 0},
{value: &s.DistributionMode, omit: s.DistributionMode == ""},
{value: s.Filter, omit: len(s.Filter) == 0},
{value: &s.DefaultOutcome, omit: s.DefaultOutcome == nil},
{value: &s.Outcomes, omit: len(s.Outcomes) == 0},
{value: &s.Capabilities, omit: len(s.Capabilities) == 0},
})
}
func (s *source) unmarshal(r *buffer) error {
return unmarshalComposite(r, typeCodeSource, []unmarshalField{
{field: &s.Address},
{field: &s.Durable},
{field: &s.ExpiryPolicy, handleNull: func() error { s.ExpiryPolicy = ExpirySessionEnd; return nil }},
{field: &s.Timeout},
{field: &s.Dynamic},
{field: &s.DynamicNodeProperties},
{field: &s.DistributionMode},
{field: &s.Filter},
{field: &s.DefaultOutcome},
{field: &s.Outcomes},
{field: &s.Capabilities},
}...)
}
func (s source) String() string {
return fmt.Sprintf("source{Address: %s, Durable: %d, ExpiryPolicy: %s, Timeout: %d, "+
"Dynamic: %t, DynamicNodeProperties: %v, DistributionMode: %s, Filter: %v, DefaultOutcome: %v"+
"Outcomes: %v, Capabilities: %v}",
s.Address,
s.Durable,
s.ExpiryPolicy,
s.Timeout,
s.Dynamic,
s.DynamicNodeProperties,
s.DistributionMode,
s.Filter,
s.DefaultOutcome,
s.Outcomes,
s.Capabilities,
)
}
/*
<type name="target" class="composite" source="list" provides="target">
<descriptor name="amqp:target:list" code="0x00000000:0x00000029"/>
<field name="address" type="*" requires="address"/>
<field name="durable" type="terminus-durability" default="none"/>
<field name="expiry-policy" type="terminus-expiry-policy" default="session-end"/>
<field name="timeout" type="seconds" default="0"/>
<field name="dynamic" type="boolean" default="false"/>
<field name="dynamic-node-properties" type="node-properties"/>
<field name="capabilities" type="symbol" multiple="true"/>
</type>
*/
type target struct {
// the address of the target
//
// The address of the target MUST NOT be set when sent on a attach frame sent by
// the sending link endpoint where the dynamic flag is set to true (that is where
// the sender is requesting the receiver to create an addressable node).
//
// The address of the source MUST be set when sent on a attach frame sent by the
// receiving link endpoint where the dynamic flag is set to true (that is where
// the receiver has created an addressable node at the request of the sender and
// is now communicating the address of that created node). The generated name of
// the address SHOULD include the link name and the container-id of the remote
// container to allow for ease of identification.
Address string
// indicates the durability of the terminus
//
// Indicates what state of the terminus will be retained durably: the state of durable
// messages, only existence and configuration of the terminus, or no state at all.
//
// 0: none
// 1: configuration
// 2: unsettled-state
Durable Durability
// the expiry policy of the target
//
// link-detach: The expiry timer starts when terminus is detached.
// session-end: The expiry timer starts when the most recently associated session is
// ended.
// connection-close: The expiry timer starts when most recently associated connection
// is closed.
// never: The terminus never expires.
ExpiryPolicy ExpiryPolicy
// duration that an expiring target will be retained
//
// The target starts expiring as indicated by the expiry-policy.
Timeout uint32 // seconds
// request dynamic creation of a remote node
//
// When set to true by the sending link endpoint, this field constitutes a request
// for the receiving peer to dynamically create a node at the target. In this case
// the address field MUST NOT be set.
//
// When set to true by the receiving link endpoint this field indicates creation of
// a dynamically created node. In this case the address field will contain the
// address of the created node. The generated address SHOULD include the link name
// and other available information on the initiator of the request (such as the
// remote container-id) in some recognizable form for ease of traceability.
Dynamic bool
// properties of the dynamically created node
//
// If the dynamic field is not set to true this field MUST be left unset.
//
// When set by the sending link endpoint, this field contains the desired
// properties of the node the sender wishes to be created. When set by the
// receiving link endpoint this field contains the actual properties of the
// dynamically created node. See subsection 3.5.9 for standard node properties.
// http://www.amqp.org/specification/1.0/node-properties
//
// lifetime-policy: The lifetime of a dynamically generated node.
// Definitionally, the lifetime will never be less than the lifetime
// of the link which caused its creation, however it is possible to
// extend the lifetime of dynamically created node using a lifetime
// policy. The value of this entry MUST be of a type which provides
// the lifetime-policy archetype. The following standard
// lifetime-policies are defined below: delete-on-close,
// delete-on-no-links, delete-on-no-messages or
// delete-on-no-links-or-messages.
// supported-dist-modes: The distribution modes that the node supports.
// The value of this entry MUST be one or more symbols which are valid
// distribution-modes. That is, the value MUST be of the same type as
// would be valid in a field defined with the following attributes:
// type="symbol" multiple="true" requires="distribution-mode"
DynamicNodeProperties map[symbol]interface{} // TODO: implement custom type with validation
// the extension capabilities the sender supports/desires
//
// http://www.amqp.org/specification/1.0/target-capabilities
Capabilities multiSymbol
}
func (t *target) marshal(wr *buffer) error {
return marshalComposite(wr, typeCodeTarget, []marshalField{
{value: &t.Address, omit: t.Address == ""},
{value: &t.Durable, omit: t.Durable == DurabilityNone},
{value: &t.ExpiryPolicy, omit: t.ExpiryPolicy == "" || t.ExpiryPolicy == ExpirySessionEnd},
{value: &t.Timeout, omit: t.Timeout == 0},
{value: &t.Dynamic, omit: !t.Dynamic},
{value: t.DynamicNodeProperties, omit: len(t.DynamicNodeProperties) == 0},
{value: &t.Capabilities, omit: len(t.Capabilities) == 0},
})
}
func (t *target) unmarshal(r *buffer) error {
return unmarshalComposite(r, typeCodeTarget, []unmarshalField{
{field: &t.Address},
{field: &t.Durable},
{field: &t.ExpiryPolicy, handleNull: func() error { t.ExpiryPolicy = ExpirySessionEnd; return nil }},
{field: &t.Timeout},
{field: &t.Dynamic},
{field: &t.DynamicNodeProperties},
{field: &t.Capabilities},
}...)
}
func (t target) String() string {
return fmt.Sprintf("source{Address: %s, Durable: %d, ExpiryPolicy: %s, Timeout: %d, "+
"Dynamic: %t, DynamicNodeProperties: %v, Capabilities: %v}",
t.Address,
t.Durable,
t.ExpiryPolicy,
t.Timeout,
t.Dynamic,
t.DynamicNodeProperties,
t.Capabilities,
)
}
/*
<type name="flow" class="composite" source="list" provides="frame">
<descriptor name="amqp:flow:list" code="0x00000000:0x00000013"/>
<field name="next-incoming-id" type="transfer-number"/>
<field name="incoming-window" type="uint" mandatory="true"/>
<field name="next-outgoing-id" type="transfer-number" mandatory="true"/>
<field name="outgoing-window" type="uint" mandatory="true"/>
<field name="handle" type="handle"/>
<field name="delivery-count" type="sequence-no"/>
<field name="link-credit" type="uint"/>
<field name="available" type="uint"/>
<field name="drain" type="boolean" default="false"/>
<field name="echo" type="boolean" default="false"/>
<field name="properties" type="fields"/>
</type>
*/
type performFlow struct {
// Identifies the expected transfer-id of the next incoming transfer frame.
// This value MUST be set if the peer has received the begin frame for the
// session, and MUST NOT be set if it has not. See subsection 2.5.6 for more details.
NextIncomingID *uint32 // sequence number
// Defines the maximum number of incoming transfer frames that the endpoint
// can currently receive. See subsection 2.5.6 for more details.
IncomingWindow uint32 // required
// The transfer-id that will be assigned to the next outgoing transfer frame.
// See subsection 2.5.6 for more details.
NextOutgoingID uint32 // sequence number
// Defines the maximum number of outgoing transfer frames that the endpoint
// could potentially currently send, if it was not constrained by restrictions
// imposed by its peer's incoming-window. See subsection 2.5.6 for more details.
OutgoingWindow uint32
// If set, indicates that the flow frame carries flow state information for the local
// link endpoint associated with the given handle. If not set, the flow frame is