You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@qpid.apache.org by ac...@apache.org on 2015/04/10 21:52:27 UTC
[4/5] qpid-proton git commit: PROTON-827: Initial work on Go language
client for proton.
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/d6f1b837/proton-c/bindings/go/src/qpid.apache.org/proton/interop_test.go
----------------------------------------------------------------------
diff --git a/proton-c/bindings/go/src/qpid.apache.org/proton/interop_test.go b/proton-c/bindings/go/src/qpid.apache.org/proton/interop_test.go
index c51d74a..d1703db 100644
--- a/proton-c/bindings/go/src/qpid.apache.org/proton/interop_test.go
+++ b/proton-c/bindings/go/src/qpid.apache.org/proton/interop_test.go
@@ -24,6 +24,7 @@ package proton
import (
"bytes"
+ "fmt"
"io"
"io/ioutil"
"os"
@@ -67,7 +68,7 @@ func assertDecode(d *Decoder, want interface{}, gotPtr interface{}) {
assertEqual(want, got)
// Try round trip encoding
- bytes, err := Marshal(want)
+ bytes, err := Marshal(want, nil)
assertNil(err)
n, err := Unmarshal(bytes, gotPtr)
assertNil(err)
@@ -149,7 +150,7 @@ func assertDecodeInterface(d *Decoder, want interface{}) {
assertEqual(want, got)
// Try round trip encoding
- bytes, err := Marshal(got)
+ bytes, err := Marshal(got, nil)
assertNil(err)
n, err := Unmarshal(bytes, &got2)
assertNil(err)
@@ -238,21 +239,21 @@ func TestEncodeDecode(t *testing.T) {
buf := bytes.Buffer{}
e := NewEncoder(&buf)
- e.Encode(in.s)
- e.Encode(in.i)
- e.Encode(in.u8)
- e.Encode(in.b)
- e.Encode(in.f)
- e.Encode(in.v)
+ assertNil(e.Encode(in.s))
+ assertNil(e.Encode(in.i))
+ assertNil(e.Encode(in.u8))
+ assertNil(e.Encode(in.b))
+ assertNil(e.Encode(in.f))
+ assertNil(e.Encode(in.v))
var out data
d := NewDecoder(&buf)
- d.Decode(&out.s)
- d.Decode(&out.i)
- d.Decode(&out.u8)
- d.Decode(&out.b)
- d.Decode(&out.f)
- d.Decode(&out.v)
+ assertNil(d.Decode(&out.s))
+ assertNil(d.Decode(&out.i))
+ assertNil(d.Decode(&out.u8))
+ assertNil(d.Decode(&out.b))
+ assertNil(d.Decode(&out.f))
+ assertNil(d.Decode(&out.v))
assertEqual(in, out)
}
@@ -275,7 +276,7 @@ func TestMap(t *testing.T) {
// Round trip a nested map
m = Map{int64(1): "one", "two": int32(2), true: Map{uint8(1): true, uint8(2): false}}
- bytes, err := Marshal(m)
+ bytes, err := Marshal(m, nil)
assertNil(err)
_, err = Unmarshal(bytes, &i)
assertNil(err)
@@ -288,3 +289,20 @@ func TestList(t *testing.T) {
assertDecode(d, List{int32(32), "foo", true}, &l)
assertDecode(d, List{}, &l)
}
+
+func FIXMETestMessage(t *testing.T) {
+ // FIXME aconway 2015-04-09: integrate Message encoding under marshal/unmarshal API.
+ bytes, err := ioutil.ReadAll(getReader("message"))
+ assertNil(err)
+ m, err := DecodeMessage(bytes)
+ assertNil(err)
+ fmt.Printf("%+v\n", m)
+ assertEqual(m.Body(), "hello")
+
+ bytes2 := make([]byte, len(bytes))
+ bytes2, err = m.Encode(bytes2)
+ assertNil(err)
+ assertEqual(bytes, bytes2)
+}
+
+// FIXME aconway 2015-03-13: finish the full interop test
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/d6f1b837/proton-c/bindings/go/src/qpid.apache.org/proton/marshal.go
----------------------------------------------------------------------
diff --git a/proton-c/bindings/go/src/qpid.apache.org/proton/marshal.go b/proton-c/bindings/go/src/qpid.apache.org/proton/marshal.go
index fcd414c..74bd8e2 100644
--- a/proton-c/bindings/go/src/qpid.apache.org/proton/marshal.go
+++ b/proton-c/bindings/go/src/qpid.apache.org/proton/marshal.go
@@ -28,10 +28,11 @@ import (
"unsafe"
)
-const minEncode = 256
-
/*
-Marshal encodes a value as AMQP.
+Marshal encodes a Go value as AMQP data in buffer.
+If buffer is nil, or is not large enough, a new buffer is created.
+
+Returns the buffer used for encoding with len() adjusted to the actual size of data.
Go types are encoded as follows
@@ -48,9 +49,13 @@ Go types are encoded as follows
+-------------------------------------+--------------------------------------------+
|string |string |
+-------------------------------------+--------------------------------------------+
- |[]byte |binary |
+ |[]byte, Binary |binary |
+-------------------------------------+--------------------------------------------+
- |interface{} |as to the contained type |
+ |Symbol |symbol |
+ +-------------------------------------+--------------------------------------------+
+ |interface{} |the contained type |
+ +-------------------------------------+--------------------------------------------+
+ |nil |null |
+-------------------------------------+--------------------------------------------+
|map[K]T |map with K and T converted as above |
+-------------------------------------+--------------------------------------------+
@@ -61,43 +66,56 @@ Go types are encoded as follows
|List |list, may have mixed types values |
+-------------------------------------+--------------------------------------------+
-TODO types
-
-Go: array, slice, struct
-
-Go types that cannot be marshaled
-
-complex64/128, uintptr, function, interface, channel
+TODO Go types: array, slice, struct
+Go types that cannot be marshaled: complex64/128, uintptr, function, interface, channel
*/
-func Marshal(v interface{}) (bytes []byte, err error) {
- return marshal(make([]byte, minEncode), v)
-}
-
-func marshal(bytesIn []byte, v interface{}) (bytes []byte, err error) {
+func Marshal(v interface{}, buffer []byte) (outbuf []byte, err error) {
defer doRecover(&err)
data := C.pn_data(0)
defer C.pn_data_free(data)
put(data, v)
- // FIXME aconway 2015-03-11: get size from proton.
- bytes = bytesIn
- for {
- n := int(C.pn_data_encode(data, (*C.char)(unsafe.Pointer(&bytes[0])), C.size_t(cap(bytes))))
- if n != int(C.PN_EOS) {
- if n < 0 {
- err = errorf(pnErrorName(n))
- } else {
- bytes = bytes[0:n]
- }
- return
+ encode := func(buf []byte) ([]byte, error) {
+ n := int(C.pn_data_encode(data, cPtr(buf), cLen(buf)))
+ switch {
+ case n == int(C.PN_OVERFLOW):
+ return buf, overflow
+ case n < 0:
+ return buf, dataError("marshal error", data)
+ default:
+ return buf[:n], nil
}
- bytes = make([]byte, cap(bytes)*2)
}
- return
+ return encodeGrow(buffer, encode)
+}
+
+const minEncode = 256
+
+// overflow is returned when an encoding function can't fit data in the buffer.
+var overflow = errorf("buffer too small")
+
+// encodeFn encodes into buffer[0:len(buffer)].
+// Returns buffer with length adjusted for data encoded.
+// If buffer too small, returns overflow as error.
+type encodeFn func(buffer []byte) ([]byte, error)
+
+// encodeGrow calls encode() into buffer, if it returns overflow grows the buffer.
+// Returns the final buffer.
+func encodeGrow(buffer []byte, encode encodeFn) ([]byte, error) {
+ if buffer == nil || len(buffer) == 0 {
+ buffer = make([]byte, minEncode)
+ }
+ var err error
+ for buffer, err = encode(buffer); err == overflow; buffer, err = encode(buffer) {
+ buffer = make([]byte, 2*len(buffer))
+ }
+ return buffer, err
}
func put(data *C.pn_data_t, v interface{}) {
switch v := v.(type) {
+ case nil:
+ C.pn_data_put_null(data)
case bool:
C.pn_data_put_bool(data, C.bool(v))
case int8:
@@ -133,9 +151,13 @@ func put(data *C.pn_data_t, v interface{}) {
case float64:
C.pn_data_put_double(data, C.double(v))
case string:
- C.pn_data_put_string(data, toPnBytes([]byte(v)))
+ C.pn_data_put_string(data, pnBytes([]byte(v)))
case []byte:
- C.pn_data_put_binary(data, toPnBytes(v))
+ C.pn_data_put_binary(data, pnBytes(v))
+ case Binary:
+ C.pn_data_put_binary(data, pnBytes([]byte(v)))
+ case Symbol:
+ C.pn_data_put_symbol(data, pnBytes([]byte(v)))
case Map: // Special map type
C.pn_data_put_map(data)
C.pn_data_enter(data)
@@ -154,9 +176,9 @@ func put(data *C.pn_data_t, v interface{}) {
panic(errorf("cannot marshal %s to AMQP", reflect.TypeOf(v)))
}
}
- err := pnDataError(data)
- if err != "" {
- panic(errorf("marshal %s", err))
+ err := dataError("marshal", data)
+ if err != nil {
+ panic(err)
}
return
}
@@ -194,17 +216,14 @@ func NewEncoder(w io.Writer) *Encoder {
}
func (e *Encoder) Encode(v interface{}) (err error) {
- e.buffer, err = marshal(e.buffer, v)
+ e.buffer, err = Marshal(v, e.buffer)
if err == nil {
e.writer.Write(e.buffer)
}
- return
+ return err
}
-func toPnBytes(b []byte) C.pn_bytes_t {
- if len(b) == 0 {
- return C.pn_bytes_t{0, nil}
- } else {
- return C.pn_bytes_t{C.size_t(len(b)), (*C.char)(unsafe.Pointer(&b[0]))}
- }
+func replace(data *C.pn_data_t, v interface{}) {
+ C.pn_data_clear(data)
+ put(data, v)
}
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/d6f1b837/proton-c/bindings/go/src/qpid.apache.org/proton/message.go
----------------------------------------------------------------------
diff --git a/proton-c/bindings/go/src/qpid.apache.org/proton/message.go b/proton-c/bindings/go/src/qpid.apache.org/proton/message.go
new file mode 100644
index 0000000..0788e11
--- /dev/null
+++ b/proton-c/bindings/go/src/qpid.apache.org/proton/message.go
@@ -0,0 +1,388 @@
+/*
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+*/
+
+package proton
+
+// #include <proton/types.h>
+// #include <proton/message.h>
+// #include <proton/codec.h>
+import "C"
+
+import (
+ "qpid.apache.org/proton/event"
+ "sync/atomic"
+ "time"
+)
+
+// Message is the interface to an AMQP message.
+type Message interface {
+ /**
+ * Inferred indicates how the message content
+ * is encoded into AMQP sections. If inferred is true then binary and
+ * list values in the body of the message will be encoded as AMQP DATA
+ * and AMQP SEQUENCE sections, respectively. If inferred is false,
+ * then all values in the body of the message will be encoded as AMQP
+ * VALUE sections regardless of their type.
+ */
+ Inferred() bool
+ SetInferred(bool)
+
+ /**
+ * Durable indicates that any parties taking responsibility
+ * for the message must durably store the content.
+ */
+ Durable() bool
+ SetDurable(bool)
+
+ /**
+ * Priority impacts ordering guarantees. Within a
+ * given ordered context, higher priority messages may jump ahead of
+ * lower priority messages.
+ */
+ Priority() uint8
+ SetPriority(uint8)
+
+ /**
+ * TTL or Time To Live, a message it may be dropped after this duration
+ */
+ TTL() time.Duration
+ SetTTL(time.Duration)
+
+ /**
+ * FirstAcquirer indicates
+ * that the recipient of the message is the first recipient to acquire
+ * the message, i.e. there have been no failed delivery attempts to
+ * other acquirers. Note that this does not mean the message has not
+ * been delivered to, but not acquired, by other recipients.
+ */
+ FirstAcquirer() bool
+ SetFirstAcquirer(bool)
+
+ /**
+ * DeliveryCount tracks how many attempts have been made to
+ * delivery a message.
+ */
+ DeliveryCount() uint32
+ SetDeliveryCount(uint32)
+
+ /**
+ * MessageId provides a unique identifier for a message.
+ * it can be an a string, an unsigned long, a uuid or a
+ * binary value.
+ */
+ MessageId() interface{}
+ SetMessageId(interface{})
+
+ UserId() string
+ SetUserId(string)
+
+ Address() string
+ SetAddress(string)
+
+ Subject() string
+ SetSubject(string)
+
+ ReplyTo() string
+ SetReplyTo(string)
+
+ /**
+ * CorrelationId is set on correlated request and response messages. It can be an a string, an unsigned long, a uuid or a
+ * binary value.
+ */
+ CorrelationId() interface{}
+ SetCorrelationId(interface{})
+
+ ContentType() string
+ SetContentType(string)
+
+ ContentEncoding() string
+ SetContentEncoding(string)
+
+ // ExpiryTime indicates an absoulte time when the message may be dropped.
+ // A Zero time (i.e. t.isZero() == true) indicates a message never expires.
+ ExpiryTime() time.Time
+ SetExpiryTime(time.Time)
+
+ CreationTime() time.Time
+ SetCreationTime(time.Time)
+
+ GroupId() string
+ SetGroupId(string)
+
+ GroupSequence() int32
+ SetGroupSequence(int32)
+
+ ReplyToGroupId() string
+ SetReplyToGroupId(string)
+
+ /**
+ * Instructions can be used to access or modify AMQP delivery instructions.
+ */
+ Instructions() *map[string]interface{}
+
+ /**
+ * Annotations can be used to access or modify AMQP annotations.
+ */
+ Annotations() *map[string]interface{}
+
+ /**
+ * Properties can be used to access or modify the application properties of a message.
+ */
+ Properties() *map[string]interface{}
+
+ /**
+ * Body of the message can be any AMQP encodable type.
+ */
+ Body() interface{}
+ SetBody(interface{})
+
+ // Encode encodes the message as AMQP data. If buffer is non-nil and is large enough
+ // the message is encoded into it, otherwise a new buffer is created.
+ // Returns the buffer containing the message.
+ Encode(buffer []byte) ([]byte, error)
+
+ // Send the message over an outgoing link
+ Send(event.Link) (event.Delivery, error)
+}
+
+func NewMessage() Message {
+ pn := C.pn_message() // Pick up default setting from C message.
+ defer C.pn_message_free(pn)
+ return goMessage(pn)
+}
+
+// Message implementation copies all message data into Go space so it can be proprely
+// memory managed.
+//
+type message struct {
+ inferred, durable, firstAcquirer bool
+ priority uint8
+ ttl time.Duration
+ deliveryCount uint32
+ messageId interface{}
+ userId, address, subject, replyTo string
+ contentType, contentEncoding string
+ groupId, replyToGroupId string
+ creationTime, expiryTime time.Time
+ groupSequence int32
+ correlationId interface{}
+ instructions, annotations, properties map[string]interface{}
+ body interface{}
+}
+
+func (m *message) Inferred() bool { return m.inferred }
+func (m *message) SetInferred(b bool) { m.inferred = b }
+func (m *message) Durable() bool { return m.durable }
+func (m *message) SetDurable(b bool) { m.durable = b }
+func (m *message) Priority() uint8 { return m.priority }
+func (m *message) SetPriority(b uint8) { m.priority = b }
+func (m *message) TTL() time.Duration { return m.ttl }
+func (m *message) SetTTL(d time.Duration) { m.ttl = d }
+func (m *message) FirstAcquirer() bool { return m.firstAcquirer }
+func (m *message) SetFirstAcquirer(b bool) { m.firstAcquirer = b }
+func (m *message) DeliveryCount() uint32 { return m.deliveryCount }
+func (m *message) SetDeliveryCount(c uint32) { m.deliveryCount = c }
+func (m *message) MessageId() interface{} { return m.messageId }
+func (m *message) SetMessageId(id interface{}) { m.messageId = id }
+func (m *message) UserId() string { return m.userId }
+func (m *message) SetUserId(s string) { m.userId = s }
+func (m *message) Address() string { return m.address }
+func (m *message) SetAddress(s string) { m.address = s }
+func (m *message) Subject() string { return m.subject }
+func (m *message) SetSubject(s string) { m.subject = s }
+func (m *message) ReplyTo() string { return m.replyTo }
+func (m *message) SetReplyTo(s string) { m.replyTo = s }
+func (m *message) CorrelationId() interface{} { return m.correlationId }
+func (m *message) SetCorrelationId(c interface{}) { m.correlationId = c }
+func (m *message) ContentType() string { return m.contentType }
+func (m *message) SetContentType(s string) { m.contentType = s }
+func (m *message) ContentEncoding() string { return m.contentEncoding }
+func (m *message) SetContentEncoding(s string) { m.contentEncoding = s }
+func (m *message) ExpiryTime() time.Time { return m.expiryTime }
+func (m *message) SetExpiryTime(t time.Time) { m.expiryTime = t }
+func (m *message) CreationTime() time.Time { return m.creationTime }
+func (m *message) SetCreationTime(t time.Time) { m.creationTime = t }
+func (m *message) GroupId() string { return m.groupId }
+func (m *message) SetGroupId(s string) { m.groupId = s }
+func (m *message) GroupSequence() int32 { return m.groupSequence }
+func (m *message) SetGroupSequence(s int32) { m.groupSequence = s }
+func (m *message) ReplyToGroupId() string { return m.replyToGroupId }
+func (m *message) SetReplyToGroupId(s string) { m.replyToGroupId = s }
+func (m *message) Instructions() *map[string]interface{} { return &m.instructions }
+func (m *message) Annotations() *map[string]interface{} { return &m.annotations }
+func (m *message) Properties() *map[string]interface{} { return &m.properties }
+func (m *message) Body() interface{} { return m.body }
+func (m *message) SetBody(b interface{}) { m.body = b }
+
+// rewindGet rewinds and then gets the value from a data object.
+func rewindGet(data *C.pn_data_t, v interface{}) {
+ if data != nil && C.pn_data_size(data) > 0 {
+ C.pn_data_rewind(data)
+ C.pn_data_next(data)
+ get(data, v)
+ }
+}
+
+// goMessage populates a Go message from a pn_message_t
+func goMessage(pn *C.pn_message_t) *message {
+ m := &message{
+ inferred: bool(C.pn_message_is_inferred(pn)),
+ durable: bool(C.pn_message_is_durable(pn)),
+ priority: uint8(C.pn_message_get_priority(pn)),
+ ttl: time.Duration(C.pn_message_get_ttl(pn)) * time.Millisecond,
+ firstAcquirer: bool(C.pn_message_is_first_acquirer(pn)),
+ deliveryCount: uint32(C.pn_message_get_delivery_count(pn)),
+ userId: goString(C.pn_message_get_user_id(pn)),
+ address: C.GoString(C.pn_message_get_address(pn)),
+ subject: C.GoString(C.pn_message_get_subject(pn)),
+ replyTo: C.GoString(C.pn_message_get_reply_to(pn)),
+ contentType: C.GoString(C.pn_message_get_content_type(pn)),
+ contentEncoding: C.GoString(C.pn_message_get_content_encoding(pn)),
+ expiryTime: time.Unix(0, int64(time.Millisecond*time.Duration(C.pn_message_get_expiry_time(pn)))),
+ creationTime: time.Unix(0, int64(time.Millisecond)*int64(C.pn_message_get_creation_time(pn))),
+ groupId: C.GoString(C.pn_message_get_group_id(pn)),
+ groupSequence: int32(C.pn_message_get_group_sequence(pn)),
+ replyToGroupId: C.GoString(C.pn_message_get_reply_to_group_id(pn)),
+ messageId: nil,
+ correlationId: nil,
+ instructions: make(map[string]interface{}),
+ annotations: make(map[string]interface{}),
+ properties: make(map[string]interface{}),
+ }
+ rewindGet(C.pn_message_id(pn), &m.messageId)
+ rewindGet(C.pn_message_correlation_id(pn), &m.correlationId)
+ rewindGet(C.pn_message_instructions(pn), &m.instructions)
+ rewindGet(C.pn_message_annotations(pn), &m.annotations)
+ rewindGet(C.pn_message_properties(pn), &m.properties)
+ rewindGet(C.pn_message_body(pn), &m.body)
+ return m
+}
+
+// pnMessage populates a pn_message_t from a Go message.
+func (m *message) pnMessage() *C.pn_message_t {
+ pn := C.pn_message()
+ C.pn_message_set_inferred(pn, C.bool(m.Inferred()))
+ C.pn_message_set_durable(pn, C.bool(m.Durable()))
+ C.pn_message_set_priority(pn, C.uint8_t(m.priority))
+ C.pn_message_set_ttl(pn, C.pn_millis_t(m.TTL()/time.Millisecond))
+ C.pn_message_set_first_acquirer(pn, C.bool(m.FirstAcquirer()))
+ C.pn_message_set_delivery_count(pn, C.uint32_t(m.deliveryCount))
+ replace(C.pn_message_id(pn), m.MessageId())
+ C.pn_message_set_user_id(pn, pnBytes([]byte(m.UserId())))
+ C.pn_message_set_address(pn, C.CString(m.Address()))
+ C.pn_message_set_subject(pn, C.CString(m.Subject()))
+ C.pn_message_set_reply_to(pn, C.CString(m.ReplyTo()))
+ replace(C.pn_message_correlation_id(pn), m.CorrelationId())
+ C.pn_message_set_content_type(pn, C.CString(m.ContentType()))
+ C.pn_message_set_content_encoding(pn, C.CString(m.ContentEncoding()))
+ C.pn_message_set_expiry_time(pn, pnTime(m.ExpiryTime()))
+ C.pn_message_set_creation_time(pn, pnTime(m.CreationTime()))
+ C.pn_message_set_group_id(pn, C.CString(m.GroupId()))
+ C.pn_message_set_group_sequence(pn, C.pn_sequence_t(m.GroupSequence()))
+ C.pn_message_set_reply_to_group_id(pn, C.CString(m.ReplyToGroupId()))
+ replace(C.pn_message_instructions(pn), *m.Instructions())
+ replace(C.pn_message_annotations(pn), *m.Annotations())
+ replace(C.pn_message_properties(pn), *m.Properties())
+ replace(C.pn_message_body(pn), m.Body())
+ return pn
+}
+
+// FIXME aconway 2015-04-08: Move message encode/decode under Marshal/Unmarshal interfaces.
+
+// DecodeMessage decodes bytes as a message
+func DecodeMessage(data []byte) (Message, error) {
+ pnMsg := C.pn_message()
+ defer C.pn_message_free(pnMsg)
+ if len(data) == 0 {
+ return nil, errorf("empty buffer for decode")
+ }
+ if C.pn_message_decode(pnMsg, cPtr(data), cLen(data)) < 0 {
+ return nil, pnError("decoding message", C.pn_message_error(pnMsg))
+ }
+ return goMessage(pnMsg), nil
+}
+
+// EventMessage decodes the message containined in a delivery event.
+func EventMessage(e event.Event) (Message, error) {
+ // FIXME aconway 2015-04-07: temporary
+ // defer doRecover(&err)
+ delivery := e.Delivery()
+ if !delivery.Readable() || delivery.Partial() {
+ return nil, errorf("attempting to get incomplete message")
+ }
+ data := make([]byte, delivery.Pending())
+ result := delivery.Link().Recv(data)
+ if result != len(data) {
+ return nil, errorf("cannot receive message: %s", pnErrorName(result))
+ }
+ return DecodeMessage(data)
+}
+
+// Encode the message into bufffer.
+// If buffer is nil or len(buffer) is not sufficient to encode the message a larger
+// buffer will be returned.
+func (m *message) Encode(buffer []byte) ([]byte, error) {
+ pn := m.pnMessage()
+ defer C.pn_message_free(pn)
+ encode := func(buf []byte) ([]byte, error) {
+ len := cLen(buf)
+ result := C.pn_message_encode(pn, cPtr(buf), &len)
+ switch {
+ case result == C.PN_OVERFLOW:
+ return buf, overflow
+ case result < 0:
+ return buf, errorf("cannot encode message: %s", pnErrorName(int(result)))
+ default:
+ return buf[:len], nil
+ }
+ }
+ return encodeGrow(buffer, encode)
+}
+
+// FIXME aconway 2015-04-08: proper handling of delivery tags.
+var tag uint64
+
+func getTag() string {
+ return string(atomic.AddUint64(&tag, 1))
+}
+
+func (m *message) Send(link event.Link) (event.Delivery, error) {
+ if !link.IsSender() {
+ return event.Delivery{}, errorf("attempt to send message on receiving link")
+ }
+ // FIXME aconway 2015-04-08: buffering, error handling
+ delivery := link.Delivery(getTag())
+ bytes, err := m.Encode(nil)
+ if err != nil {
+ return event.Delivery{}, errorf("cannot send mesage %s", err)
+ }
+ result := link.Send(bytes)
+ link.Advance()
+ if result != len(bytes) {
+ if result < 0 {
+ return delivery, errorf("send failed %v", pnErrorName(result))
+ } else {
+ return delivery, errorf("send incomplete %v of %v", result, len(bytes))
+ }
+ }
+ if link.RemoteSndSettleMode() == event.PnSndSettled { // FIXME aconway 2015-04-08: enum names
+ delivery.Settle()
+ }
+ return delivery, nil
+}
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/d6f1b837/proton-c/bindings/go/src/qpid.apache.org/proton/message_test.go
----------------------------------------------------------------------
diff --git a/proton-c/bindings/go/src/qpid.apache.org/proton/message_test.go b/proton-c/bindings/go/src/qpid.apache.org/proton/message_test.go
new file mode 100644
index 0000000..2baec22
--- /dev/null
+++ b/proton-c/bindings/go/src/qpid.apache.org/proton/message_test.go
@@ -0,0 +1,90 @@
+/*
+Licensed to the Apache Software Foundation (ASF) under one
+or more contributor license agreements. See the NOTICE file
+distributed with this work for additional information
+regarding copyright ownership. The ASF licenses this file
+to you under the Apache License, Version 2.0 (the
+"License"); you may not use this file except in compliance
+with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing,
+software distributed under the License is distributed on an
+"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+KIND, either express or implied. See the License for the
+specific language governing permissions and limitations
+under the License.
+*/
+
+package proton
+
+import (
+ "reflect"
+ "testing"
+ "time"
+)
+
+func roundTrip(t *testing.T, m Message) {
+ buffer, err := m.Encode(nil)
+ if err != nil {
+ t.Fatalf("Encode failed: %v", err)
+ }
+ m2, err := DecodeMessage(buffer)
+ if err != nil {
+ t.Fatalf("Decode failed: %v", err)
+ }
+ if !reflect.DeepEqual(m, m2) {
+ t.Errorf("Message mismatch got\n%#v\nwant\n%#v", m, m2)
+ }
+}
+
+func TestDefaultMessageRoundTrip(t *testing.T) {
+ m := NewMessage()
+ // Check defaults
+ assertEqual(m.Inferred(), false)
+ assertEqual(m.Durable(), false)
+ assertEqual(m.Priority(), uint8(4))
+ assertEqual(m.TTL(), time.Duration(0))
+ assertEqual(m.UserId(), "")
+ assertEqual(m.Address(), "")
+ assertEqual(m.Subject(), "")
+ assertEqual(m.ReplyTo(), "")
+ assertEqual(m.ContentType(), "")
+ assertEqual(m.ContentEncoding(), "")
+ assertEqual(m.GroupId(), "")
+ assertEqual(m.GroupSequence(), int32(0))
+ assertEqual(m.ReplyToGroupId(), "")
+ assertEqual(m.MessageId(), nil)
+ assertEqual(m.CorrelationId(), nil)
+ assertEqual(*m.Instructions(), map[string]interface{}{})
+ assertEqual(*m.Annotations(), map[string]interface{}{})
+ assertEqual(*m.Properties(), map[string]interface{}{})
+ assertEqual(m.Body(), nil)
+
+ roundTrip(t, m)
+}
+
+func TestMessageRoundTrip(t *testing.T) {
+ m := NewMessage()
+ m.SetInferred(false)
+ m.SetDurable(true)
+ m.SetPriority(42)
+ m.SetTTL(0)
+ m.SetUserId("user")
+ m.SetAddress("address")
+ m.SetSubject("subject")
+ m.SetReplyTo("replyto")
+ m.SetContentType("content")
+ m.SetContentEncoding("encoding")
+ m.SetGroupId("group")
+ m.SetGroupSequence(42)
+ m.SetReplyToGroupId("replytogroup")
+ m.SetMessageId("id")
+ m.SetCorrelationId("correlation")
+ *m.Instructions() = map[string]interface{}{"instructions": "foo"}
+ *m.Annotations() = map[string]interface{}{"annotations": "foo"}
+ *m.Properties() = map[string]interface{}{"int": int32(32), "bool": true, "string": "foo"}
+ m.SetBody("hello")
+ roundTrip(t, m)
+}
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/d6f1b837/proton-c/bindings/go/src/qpid.apache.org/proton/types.go
----------------------------------------------------------------------
diff --git a/proton-c/bindings/go/src/qpid.apache.org/proton/types.go b/proton-c/bindings/go/src/qpid.apache.org/proton/types.go
index 561ac9a..07f8aea 100644
--- a/proton-c/bindings/go/src/qpid.apache.org/proton/types.go
+++ b/proton-c/bindings/go/src/qpid.apache.org/proton/types.go
@@ -20,90 +20,74 @@ under the License.
package proton
// #include <proton/codec.h>
+// const pn_type_t PN_DATA_TYPE_ERROR = (pn_type_t) -1;
import "C"
import (
"bytes"
"fmt"
"reflect"
+ "time"
+ "unsafe"
)
-type pnGet func(data *C.pn_data_t) reflect.Value
-type pnPut func(data *C.pn_data_t, v interface{})
-
-// Information about a proton type
-type pnType struct {
- code C.pn_type_t // AMQP type code
- name string // AMQP type name
-}
-
-func (pt *pnType) String() string { return pt.name }
-
-// pnType definitions for each proton type.
-var (
- pnNull = pnType{code: C.PN_NULL, name: "null"}
- pnBool = pnType{code: C.PN_BOOL, name: "bool"}
- pnUbyte = pnType{code: C.PN_UBYTE, name: "ubyte"}
- pnByte = pnType{code: C.PN_BYTE, name: "byte"}
- pnUshort = pnType{code: C.PN_USHORT, name: "ushort"}
- pnShort = pnType{code: C.PN_SHORT, name: "short"}
- pnChar = pnType{code: C.PN_CHAR, name: "char"}
- pnUint = pnType{code: C.PN_UINT, name: "uint"}
- pnInt = pnType{code: C.PN_INT, name: "int"}
- pnUlong = pnType{code: C.PN_ULONG, name: "ulong"}
- pnLong = pnType{code: C.PN_LONG, name: "long"}
- pnTimestamp = pnType{code: C.PN_TIMESTAMP, name: "timestamp"}
- pnFloat = pnType{code: C.PN_FLOAT, name: "float"}
- pnDouble = pnType{code: C.PN_DOUBLE, name: "double"}
- pnDecimal32 = pnType{code: C.PN_DECIMAL32, name: "decimal32"}
- pnDecimal64 = pnType{code: C.PN_DECIMAL64, name: "decimal64"}
- pnDecimal128 = pnType{code: C.PN_DECIMAL128, name: "decimal128"}
- pnUuid = pnType{code: C.PN_UUID, name: "uuid"}
- pnBinary = pnType{code: C.PN_BINARY, name: "binary"}
- pnString = pnType{code: C.PN_STRING, name: "string"}
- pnSymbol = pnType{code: C.PN_SYMBOL, name: "symbol"}
- pnDescribed = pnType{code: C.PN_DESCRIBED, name: "described"}
- pnArray = pnType{code: C.PN_ARRAY, name: "array"}
- pnList = pnType{code: C.PN_LIST, name: "list"}
- pnMap = pnType{code: C.PN_MAP, name: "map"}
-)
-
-// Map from pn_type_t codes to pnType structs
-var pnTypes = map[C.pn_type_t]*pnType{
- C.PN_NULL: &pnNull,
- C.PN_BOOL: &pnBool,
- C.PN_UBYTE: &pnUbyte,
- C.PN_BYTE: &pnByte,
- C.PN_USHORT: &pnUshort,
- C.PN_SHORT: &pnShort,
- C.PN_UINT: &pnUint,
- C.PN_INT: &pnInt,
- C.PN_CHAR: &pnChar,
- C.PN_ULONG: &pnUlong,
- C.PN_LONG: &pnLong,
- C.PN_TIMESTAMP: &pnTimestamp,
- C.PN_FLOAT: &pnFloat,
- C.PN_DOUBLE: &pnDouble,
- C.PN_DECIMAL32: &pnDecimal32,
- C.PN_DECIMAL64: &pnDecimal64,
- C.PN_DECIMAL128: &pnDecimal128,
- C.PN_UUID: &pnUuid,
- C.PN_BINARY: &pnBinary,
- C.PN_STRING: &pnString,
- C.PN_SYMBOL: &pnSymbol,
- C.PN_DESCRIBED: &pnDescribed,
- C.PN_ARRAY: &pnArray,
- C.PN_LIST: &pnList,
- C.PN_MAP: &pnMap,
-}
-
-// Get the pnType for a pn_type_t code, panic if unknown code.
-func getPnType(code C.pn_type_t) *pnType {
- pt, ok := pnTypes[code]
- if !ok {
- panic(errorf("unknown AMQP type code %v", code))
+func pnTypeString(pt C.pn_type_t) string {
+ switch pt {
+ case C.PN_NULL:
+ return "null"
+ case C.PN_BOOL:
+ return "bool"
+ case C.PN_UBYTE:
+ return "ubyte"
+ case C.PN_BYTE:
+ return "byte"
+ case C.PN_USHORT:
+ return "ushort"
+ case C.PN_SHORT:
+ return "short"
+ case C.PN_CHAR:
+ return "char"
+ case C.PN_UINT:
+ return "uint"
+ case C.PN_INT:
+ return "int"
+ case C.PN_ULONG:
+ return "ulong"
+ case C.PN_LONG:
+ return "long"
+ case C.PN_TIMESTAMP:
+ return "timestamp"
+ case C.PN_FLOAT:
+ return "float"
+ case C.PN_DOUBLE:
+ return "double"
+ case C.PN_DECIMAL32:
+ return "decimal32"
+ case C.PN_DECIMAL64:
+ return "decimal64"
+ case C.PN_DECIMAL128:
+ return "decimal128"
+ case C.PN_UUID:
+ return "uuid"
+ case C.PN_BINARY:
+ return "binary"
+ case C.PN_STRING:
+ return "string"
+ case C.PN_SYMBOL:
+ return "symbol"
+ case C.PN_DESCRIBED:
+ return "described"
+ case C.PN_ARRAY:
+ return "array"
+ case C.PN_LIST:
+ return "list"
+ case C.PN_MAP:
+ return "map"
+ case C.PN_DATA_TYPE_ERROR:
+ return "no-data"
+ default:
+ return fmt.Sprintf("unknown-type(%d)", pt)
}
- return pt
}
// Go types
@@ -112,14 +96,23 @@ var (
valueType = reflect.TypeOf(reflect.Value{})
)
+// FIXME aconway 2015-04-08: can't handle AMQP maps with key types that are not valid Go map keys.
+
// Map is a generic map that can have mixed key and value types and so can represent any AMQP map
-//
type Map map[interface{}]interface{}
// List is a generic list that can hold mixed values and can represent any AMQP list.
//
type List []interface{}
+// Symbol is a string that is encoded as an AMQP symbol
+type Symbol string
+
+// Binary is a string that is encoded as an AMQP binary.
+// It is a string rather than a byte[] because byte[] is not hashable and can't be used as
+// a map key, AMQP frequently uses binary types as map keys. It can convert to and from []byte
+type Binary string
+
// GoString for Map prints values with their types, useful for debugging.
func (m Map) GoString() string {
out := &bytes.Buffer{}
@@ -149,3 +142,52 @@ func (l List) GoString() string {
fmt.Fprint(out, "}")
return out.String()
}
+
+// pnTime converts Go time.Time to Proton millisecond Unix time.
+func pnTime(t time.Time) C.pn_timestamp_t {
+ secs := t.Unix()
+ // Note: sub-second accuracy is not guaraunteed if the Unix time in
+ // nanoseconds cannot be represented by an int64 (sometime around year 2260)
+ msecs := (t.UnixNano() % int64(time.Second)) / int64(time.Millisecond)
+ return C.pn_timestamp_t(secs*1000 + msecs)
+}
+
+// goTime converts a pn_timestamp_t to a Go time.Time.
+func goTime(t C.pn_timestamp_t) time.Time {
+ secs := int64(t) / 1000
+ nsecs := (int64(t) % 1000) * int64(time.Millisecond)
+ return time.Unix(secs, nsecs)
+}
+
+func goBytes(cBytes C.pn_bytes_t) (bytes []byte) {
+ if cBytes.start != nil {
+ bytes = C.GoBytes(unsafe.Pointer(cBytes.start), C.int(cBytes.size))
+ }
+ return
+}
+
+func goString(cBytes C.pn_bytes_t) (str string) {
+ if cBytes.start != nil {
+ str = C.GoStringN(cBytes.start, C.int(cBytes.size))
+ }
+ return
+}
+
+func pnBytes(b []byte) C.pn_bytes_t {
+ if len(b) == 0 {
+ return C.pn_bytes_t{0, nil}
+ } else {
+ return C.pn_bytes_t{C.size_t(len(b)), (*C.char)(unsafe.Pointer(&b[0]))}
+ }
+}
+
+func cPtr(b []byte) *C.char {
+ if len(b) == 0 {
+ return nil
+ }
+ return (*C.char)(unsafe.Pointer(&b[0]))
+}
+
+func cLen(b []byte) C.size_t {
+ return C.size_t(len(b))
+}
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/d6f1b837/proton-c/bindings/go/src/qpid.apache.org/proton/unfinished.go
----------------------------------------------------------------------
diff --git a/proton-c/bindings/go/src/qpid.apache.org/proton/unfinished.go b/proton-c/bindings/go/src/qpid.apache.org/proton/unfinished.go
deleted file mode 100644
index f9ba50a..0000000
--- a/proton-c/bindings/go/src/qpid.apache.org/proton/unfinished.go
+++ /dev/null
@@ -1,53 +0,0 @@
-/*
-Licensed to the Apache Software Foundation (ASF) under one
-or more contributor license agreements. See the NOTICE file
-distributed with this work for additional information
-regarding copyright ownership. The ASF licenses this file
-to you under the Apache License, Version 2.0 (the
-"License"); you may not use this file except in compliance
-with the License. You may obtain a copy of the License at
-
- http://www.apache.org/licenses/LICENSE-2.0
-
-Unless required by applicable law or agreed to in writing,
-software distributed under the License is distributed on an
-"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
-KIND, either express or implied. See the License for the
-specific language governing permissions and limitations
-under the License.
-*/
-
-package proton
-
-import (
- "net"
-)
-
-type Acceptor struct{}
-
-type Event struct {
- Container *Container
- Message *Message
-}
-
-type Connection struct{}
-
-func (e *Event) Connection() *Connection { return nil }
-
-func (e *Event) Sender() *Sender { return nil }
-
-func (c *Connection) NewSender(name string) *Sender { return nil }
-
-func (c *Connection) Close() {}
-
-type Container struct{}
-
-type Sender struct{}
-
-func (s *Sender) Credit() int { return 0 }
-
-func (s *Sender) Send(m *Message) {}
-
-func Run(connection net.Conn, handler interface{}) error { return nil }
-
-type Message struct{ Body interface{} }
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/d6f1b837/proton-c/bindings/go/src/qpid.apache.org/proton/unmarshal.go
----------------------------------------------------------------------
diff --git a/proton-c/bindings/go/src/qpid.apache.org/proton/unmarshal.go b/proton-c/bindings/go/src/qpid.apache.org/proton/unmarshal.go
index fb09d0f..af6c933 100644
--- a/proton-c/bindings/go/src/qpid.apache.org/proton/unmarshal.go
+++ b/proton-c/bindings/go/src/qpid.apache.org/proton/unmarshal.go
@@ -29,7 +29,7 @@ import (
"unsafe"
)
-const minDecode = 256
+const minDecode = 1024
//
// Decoding from a pn_data_t
@@ -103,9 +103,11 @@ Types are converted as follows:
+---------------------------+----------------------------------------------------------------------+
|string, []byte |string, symbol or binary. |
+---------------------------+----------------------------------------------------------------------+
+ |Symbol |symbol |
+ +---------------------------+----------------------------------------------------------------------+
|map[K]T |map, provided all keys and values can unmarshal to types K, T |
+---------------------------+----------------------------------------------------------------------+
- |Map|map, any AMQP map |
+ |Map |map, any AMQP map |
+---------------------------+----------------------------------------------------------------------+
|interface{} |Any AMQP value can be unmarshaled to an interface{} as follows: |
| +------------------------+---------------------------------------------+
@@ -121,9 +123,13 @@ Types are converted as follows:
| +------------------------+---------------------------------------------+
| |float, double |float32, float64 |
| +------------------------+---------------------------------------------+
- | |string, symbol |string |
+ | |string |string |
+ | +------------------------+---------------------------------------------+
+ | |symbol |Symbol |
+ | +------------------------+---------------------------------------------+
+ | |binary |Binary |
| +------------------------+---------------------------------------------+
- | |binary |byte[] |
+ | |nulll |nil |
| +------------------------+---------------------------------------------+
| |map |Map |
| +------------------------+---------------------------------------------+
@@ -134,10 +140,12 @@ The following Go types cannot be unmarshaled: complex64/128, uintptr, function,
TODO types
-AMQP: null, timestamp, decimal32/64/128, uuid, described, array.
+AMQP: timestamp, decimal32/64/128, uuid, described, array.
Go: array, struct.
+Maps: currently we cannot unmarshal AMQP maps with unhashable key types, need an alternate
+representation for those.
*/
func Unmarshal(bytes []byte, v interface{}) (n int, err error) {
defer doRecover(&err)
@@ -178,9 +186,8 @@ func unmarshal(data *C.pn_data_t, bytes []byte, v interface{}) (n int) {
return
}
-// Get value v from data
+// get value from data into value pointed at by v.
func get(data *C.pn_data_t, v interface{}) {
-
pnType := C.pn_data_type(data)
switch v := v.(type) {
@@ -370,6 +377,22 @@ func get(data *C.pn_data_t, v interface{}) {
panic(newBadUnmarshal(pnType, v))
}
+ case *Binary:
+ switch pnType {
+ case C.PN_BINARY:
+ *v = Binary(goBytes(C.pn_data_get_binary(data)))
+ default:
+ panic(newBadUnmarshal(pnType, v))
+ }
+
+ case *Symbol:
+ switch pnType {
+ case C.PN_SYMBOL:
+ *v = Symbol(goBytes(C.pn_data_get_symbol(data)))
+ default:
+ panic(newBadUnmarshal(pnType, v))
+ }
+
case *interface{}:
getInterface(data, v)
@@ -386,9 +409,9 @@ func get(data *C.pn_data_t, v interface{}) {
panic(newBadUnmarshal(pnType, v))
}
}
- err := pnDataError(data)
- if err != "" {
- panic(errorf("unmarshal %s", err))
+ err := dataError("unmarshaling", data)
+ if err != nil {
+ panic(err)
}
return
}
@@ -398,6 +421,8 @@ func get(data *C.pn_data_t, v interface{}) {
func getInterface(data *C.pn_data_t, v *interface{}) {
pnType := C.pn_data_type(data)
switch pnType {
+ case C.PN_NULL:
+ *v = nil
case C.PN_BOOL:
*v = bool(C.pn_data_get_bool(data))
case C.PN_UBYTE:
@@ -423,11 +448,11 @@ func getInterface(data *C.pn_data_t, v *interface{}) {
case C.PN_DOUBLE:
*v = float64(C.pn_data_get_double(data))
case C.PN_BINARY:
- *v = goBytes(C.pn_data_get_binary(data))
+ *v = Binary(goBytes(C.pn_data_get_binary(data)))
case C.PN_STRING:
*v = goString(C.pn_data_get_string(data))
case C.PN_SYMBOL:
- *v = goString(C.pn_data_get_symbol(data))
+ *v = Symbol(goString(C.pn_data_get_symbol(data)))
case C.PN_MAP:
m := make(Map)
get(data, &m)
@@ -435,12 +460,13 @@ func getInterface(data *C.pn_data_t, v *interface{}) {
case C.PN_LIST:
l := make(List, 0)
get(data, &l)
- *v = l
+ *v = l // FIXME aconway 2015-03-13: avoid the copy?
default:
panic(newBadUnmarshal(pnType, v))
}
}
+// get into map pointed at by v
func getMap(data *C.pn_data_t, v interface{}) {
pnType := C.pn_data_type(data)
if pnType != C.PN_MAP {
@@ -482,6 +508,7 @@ func getList(data *C.pn_data_t, v interface{}) {
}
C.pn_data_exit(data)
}
+ // FIXME aconway 2015-04-09: avoid the copy?
reflect.ValueOf(v).Elem().Set(listValue)
}
@@ -492,26 +519,12 @@ func decode(data *C.pn_data_t, bytes []byte) int {
if len(bytes) == 0 {
return 0
}
- cBuf := (*C.char)(unsafe.Pointer(&bytes[0]))
- n := int(C.pn_data_decode(data, cBuf, C.size_t(len(bytes))))
- if n == int(C.PN_EOS) {
+ n := int(C.pn_data_decode(data, cPtr(bytes), cLen(bytes)))
+ if n == int(C.PN_UNDERFLOW) {
+ C.pn_error_clear(C.pn_data_error(data))
return 0
} else if n <= 0 {
panic(errorf("unmarshal %s", pnErrorName(n)))
}
return n
}
-
-func goBytes(cBytes C.pn_bytes_t) (bytes []byte) {
- if cBytes.start != nil {
- bytes = C.GoBytes(unsafe.Pointer(cBytes.start), C.int(cBytes.size))
- }
- return
-}
-
-func goString(cBytes C.pn_bytes_t) (str string) {
- if cBytes.start != nil {
- str = C.GoStringN(cBytes.start, C.int(cBytes.size))
- }
- return
-}
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/d6f1b837/proton-c/bindings/go/src/qpid.apache.org/proton/url.go
----------------------------------------------------------------------
diff --git a/proton-c/bindings/go/src/qpid.apache.org/proton/url.go b/proton-c/bindings/go/src/qpid.apache.org/proton/url.go
index 079dfb2..72be720 100644
--- a/proton-c/bindings/go/src/qpid.apache.org/proton/url.go
+++ b/proton-c/bindings/go/src/qpid.apache.org/proton/url.go
@@ -20,7 +20,6 @@ under the License.
package proton
/*
-#cgo LDFLAGS: -lqpid-proton
#include <stdlib.h>
#include <string.h>
#include <proton/url.h>
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org