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/10/23 16:36:09 UTC
[07/50] [abbrv] qpid-proton git commit: PROTON-1011: Go example of
event driven broker. Package renaming and some new features.
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/478ba4ea/proton-c/bindings/go/src/qpid.apache.org/proton/amqp/unmarshal.go
----------------------------------------------------------------------
diff --git a/proton-c/bindings/go/src/qpid.apache.org/proton/amqp/unmarshal.go b/proton-c/bindings/go/src/qpid.apache.org/proton/amqp/unmarshal.go
deleted file mode 100644
index d645273..0000000
--- a/proton-c/bindings/go/src/qpid.apache.org/proton/amqp/unmarshal.go
+++ /dev/null
@@ -1,556 +0,0 @@
-/*
-Licensed to the Apache Software Foundation (ASF) under one
-oor 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 amqp
-
-// #include <proton/codec.h>
-import "C"
-
-import (
- "bytes"
- "fmt"
- "io"
- "qpid.apache.org/proton/internal"
- "reflect"
- "unsafe"
-)
-
-const minDecode = 1024
-
-// Error returned if AMQP data cannot be unmarshaled as the desired Go type.
-type UnmarshalError struct {
- // The name of the AMQP type.
- AMQPType string
- // The Go type.
- GoType reflect.Type
-}
-
-func newUnmarshalError(pnType C.pn_type_t, v interface{}) *UnmarshalError {
- return &UnmarshalError{Type(pnType).String(), reflect.TypeOf(v)}
-}
-
-func (e UnmarshalError) Error() string {
- if e.GoType.Kind() != reflect.Ptr {
- return fmt.Sprintf("proton: cannot unmarshal to type %s, not a pointer", e.GoType)
- } else {
- return fmt.Sprintf("proton: cannot unmarshal AMQP %s to %s", e.AMQPType, e.GoType)
- }
-}
-
-func doRecover(err *error) {
- r := recover()
- switch r := r.(type) {
- case nil:
- case *UnmarshalError, internal.Error:
- *err = r.(error)
- default:
- panic(r)
- }
-}
-
-//
-// Decoding from a pn_data_t
-//
-// NOTE: we use panic() to signal a decoding error, simplifies decoding logic.
-// We recover() at the highest possible level - i.e. in the exported Unmarshal or Decode.
-//
-
-// Decoder decodes AMQP values from an io.Reader.
-//
-type Decoder struct {
- reader io.Reader
- buffer bytes.Buffer
-}
-
-// NewDecoder returns a new decoder that reads from r.
-//
-// The decoder has it's own buffer and may read more data than required for the
-// AMQP values requested. Use Buffered to see if there is data left in the
-// buffer.
-//
-func NewDecoder(r io.Reader) *Decoder {
- return &Decoder{r, bytes.Buffer{}}
-}
-
-// Buffered returns a reader of the data remaining in the Decoder's buffer. The
-// reader is valid until the next call to Decode.
-//
-func (d *Decoder) Buffered() io.Reader {
- return bytes.NewReader(d.buffer.Bytes())
-}
-
-// Decode reads the next AMQP value from the Reader and stores it in the value pointed to by v.
-//
-// See the documentation for Unmarshal for details about the conversion of AMQP into a Go value.
-//
-func (d *Decoder) Decode(v interface{}) (err error) {
- defer doRecover(&err)
- data := C.pn_data(0)
- defer C.pn_data_free(data)
- var n int
- for n == 0 && err == nil {
- n = decode(data, d.buffer.Bytes())
- if n == 0 { // n == 0 means not enough data, read more
- err = d.more()
- } else {
- unmarshal(v, data)
- }
- }
- d.buffer.Next(n)
- return
-}
-
-/*
-Unmarshal decodes AMQP-encoded bytes and stores the result in the value pointed to by v.
-Types are converted as follows:
-
- +---------------------------+----------------------------------------------------------------------+
- |To Go types |From AMQP types |
- +===========================+======================================================================+
- |bool |bool |
- +---------------------------+----------------------------------------------------------------------+
- |int, int8, int16, |Equivalent or smaller signed integer type: byte, short, int, long. |
- |int32, int64 | |
- +---------------------------+----------------------------------------------------------------------+
- |uint, uint8, uint16, |Equivalent or smaller unsigned integer type: ubyte, ushort, uint, |
- |uint32, uint64 types |ulong |
- +---------------------------+----------------------------------------------------------------------+
- |float32, float64 |Equivalent or smaller float or double. |
- +---------------------------+----------------------------------------------------------------------+
- |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 |
- +---------------------------+----------------------------------------------------------------------+
- |interface{} |Any AMQP value can be unmarshaled to an interface{} as follows: |
- | +------------------------+---------------------------------------------+
- | |AMQP Type |Go Type in interface{} |
- | +========================+=============================================+
- | |bool |bool |
- | +------------------------+---------------------------------------------+
- | |byte,short,int,long |int8,int16,int32,int64 |
- | +------------------------+---------------------------------------------+
- | |ubyte,ushort,uint,ulong |uint8,uint16,uint32,uint64 |
- | +------------------------+---------------------------------------------+
- | |float, double |float32, float64 |
- | +------------------------+---------------------------------------------+
- | |string |string |
- | +------------------------+---------------------------------------------+
- | |symbol |Symbol |
- | +------------------------+---------------------------------------------+
- | |binary |Binary |
- | +------------------------+---------------------------------------------+
- | |nulll |nil |
- | +------------------------+---------------------------------------------+
- | |map |Map |
- | +------------------------+---------------------------------------------+
- | |list |List |
- +---------------------------+------------------------+---------------------------------------------+
-
-The following Go types cannot be unmarshaled: uintptr, function, interface, channel.
-
-TODO
-
-Go types: array, struct.
-
-AMQP types: decimal32/64/128, char (round trip), timestamp, uuid, array, multi-section message bodies.
-
-AMQP maps with mixed/unhashable key types need an alternate representation.
-
-Described types.
-*/
-func Unmarshal(bytes []byte, v interface{}) (n int, err error) {
- defer doRecover(&err)
-
- data := C.pn_data(0)
- defer C.pn_data_free(data)
- n = decode(data, bytes)
- if n == 0 {
- err = internal.Errorf("not enough data")
- } else {
- unmarshal(v, data)
- }
- return
-}
-
-// more reads more data when we can't parse a complete AMQP type
-func (d *Decoder) more() error {
- var readSize int64 = minDecode
- if int64(d.buffer.Len()) > readSize { // Grow by doubling
- readSize = int64(d.buffer.Len())
- }
- var n int64
- n, err := d.buffer.ReadFrom(io.LimitReader(d.reader, readSize))
- if n == 0 && err == nil { // ReadFrom won't report io.EOF, just returns 0
- err = io.EOF
- }
- return err
-}
-
-// Unmarshal from data into value pointed at by v.
-func unmarshal(v interface{}, data *C.pn_data_t) {
- pnType := C.pn_data_type(data)
- switch v := v.(type) {
- case *bool:
- switch pnType {
- case C.PN_BOOL:
- *v = bool(C.pn_data_get_bool(data))
- default:
- panic(newUnmarshalError(pnType, v))
- }
- case *int8:
- switch pnType {
- case C.PN_CHAR:
- *v = int8(C.pn_data_get_char(data))
- case C.PN_BYTE:
- *v = int8(C.pn_data_get_byte(data))
- default:
- panic(newUnmarshalError(pnType, v))
- }
- case *uint8:
- switch pnType {
- case C.PN_CHAR:
- *v = uint8(C.pn_data_get_char(data))
- case C.PN_UBYTE:
- *v = uint8(C.pn_data_get_ubyte(data))
- default:
- panic(newUnmarshalError(pnType, v))
- }
- case *int16:
- switch pnType {
- case C.PN_CHAR:
- *v = int16(C.pn_data_get_char(data))
- case C.PN_BYTE:
- *v = int16(C.pn_data_get_byte(data))
- case C.PN_SHORT:
- *v = int16(C.pn_data_get_short(data))
- default:
- panic(newUnmarshalError(pnType, v))
- }
- case *uint16:
- switch pnType {
- case C.PN_CHAR:
- *v = uint16(C.pn_data_get_char(data))
- case C.PN_UBYTE:
- *v = uint16(C.pn_data_get_ubyte(data))
- case C.PN_USHORT:
- *v = uint16(C.pn_data_get_ushort(data))
- default:
- panic(newUnmarshalError(pnType, v))
- }
- case *int32:
- switch pnType {
- case C.PN_CHAR:
- *v = int32(C.pn_data_get_char(data))
- case C.PN_BYTE:
- *v = int32(C.pn_data_get_byte(data))
- case C.PN_SHORT:
- *v = int32(C.pn_data_get_short(data))
- case C.PN_INT:
- *v = int32(C.pn_data_get_int(data))
- default:
- panic(newUnmarshalError(pnType, v))
- }
- case *uint32:
- switch pnType {
- case C.PN_CHAR:
- *v = uint32(C.pn_data_get_char(data))
- case C.PN_UBYTE:
- *v = uint32(C.pn_data_get_ubyte(data))
- case C.PN_USHORT:
- *v = uint32(C.pn_data_get_ushort(data))
- case C.PN_UINT:
- *v = uint32(C.pn_data_get_uint(data))
- default:
- panic(newUnmarshalError(pnType, v))
- }
-
- case *int64:
- switch pnType {
- case C.PN_CHAR:
- *v = int64(C.pn_data_get_char(data))
- case C.PN_BYTE:
- *v = int64(C.pn_data_get_byte(data))
- case C.PN_SHORT:
- *v = int64(C.pn_data_get_short(data))
- case C.PN_INT:
- *v = int64(C.pn_data_get_int(data))
- case C.PN_LONG:
- *v = int64(C.pn_data_get_long(data))
- default:
- panic(newUnmarshalError(pnType, v))
- }
-
- case *uint64:
- switch pnType {
- case C.PN_CHAR:
- *v = uint64(C.pn_data_get_char(data))
- case C.PN_UBYTE:
- *v = uint64(C.pn_data_get_ubyte(data))
- case C.PN_USHORT:
- *v = uint64(C.pn_data_get_ushort(data))
- case C.PN_ULONG:
- *v = uint64(C.pn_data_get_ulong(data))
- default:
- panic(newUnmarshalError(pnType, v))
- }
-
- case *int:
- switch pnType {
- case C.PN_CHAR:
- *v = int(C.pn_data_get_char(data))
- case C.PN_BYTE:
- *v = int(C.pn_data_get_byte(data))
- case C.PN_SHORT:
- *v = int(C.pn_data_get_short(data))
- case C.PN_INT:
- *v = int(C.pn_data_get_int(data))
- case C.PN_LONG:
- if unsafe.Sizeof(0) == 8 {
- *v = int(C.pn_data_get_long(data))
- } else {
- panic(newUnmarshalError(pnType, v))
- }
- default:
- panic(newUnmarshalError(pnType, v))
- }
-
- case *uint:
- switch pnType {
- case C.PN_CHAR:
- *v = uint(C.pn_data_get_char(data))
- case C.PN_UBYTE:
- *v = uint(C.pn_data_get_ubyte(data))
- case C.PN_USHORT:
- *v = uint(C.pn_data_get_ushort(data))
- case C.PN_UINT:
- *v = uint(C.pn_data_get_uint(data))
- case C.PN_ULONG:
- if unsafe.Sizeof(0) == 8 {
- *v = uint(C.pn_data_get_ulong(data))
- } else {
- panic(newUnmarshalError(pnType, v))
- }
- default:
- panic(newUnmarshalError(pnType, v))
- }
-
- case *float32:
- switch pnType {
- case C.PN_FLOAT:
- *v = float32(C.pn_data_get_float(data))
- default:
- panic(newUnmarshalError(pnType, v))
- }
-
- case *float64:
- switch pnType {
- case C.PN_FLOAT:
- *v = float64(C.pn_data_get_float(data))
- case C.PN_DOUBLE:
- *v = float64(C.pn_data_get_double(data))
- default:
- panic(newUnmarshalError(pnType, v))
- }
-
- case *string:
- switch pnType {
- case C.PN_STRING:
- *v = goString(C.pn_data_get_string(data))
- case C.PN_SYMBOL:
- *v = goString(C.pn_data_get_symbol(data))
- case C.PN_BINARY:
- *v = goString(C.pn_data_get_binary(data))
- default:
- panic(newUnmarshalError(pnType, v))
- }
-
- case *[]byte:
- switch pnType {
- case C.PN_STRING:
- *v = goBytes(C.pn_data_get_string(data))
- case C.PN_SYMBOL:
- *v = goBytes(C.pn_data_get_symbol(data))
- case C.PN_BINARY:
- *v = goBytes(C.pn_data_get_binary(data))
- default:
- panic(newUnmarshalError(pnType, v))
- }
-
- case *Binary:
- switch pnType {
- case C.PN_BINARY:
- *v = Binary(goBytes(C.pn_data_get_binary(data)))
- default:
- panic(newUnmarshalError(pnType, v))
- }
-
- case *Symbol:
- switch pnType {
- case C.PN_SYMBOL:
- *v = Symbol(goBytes(C.pn_data_get_symbol(data)))
- default:
- panic(newUnmarshalError(pnType, v))
- }
-
- case *interface{}:
- getInterface(data, v)
-
- default:
- if reflect.TypeOf(v).Kind() != reflect.Ptr {
- panic(newUnmarshalError(pnType, v))
- }
- switch reflect.TypeOf(v).Elem().Kind() {
- case reflect.Map:
- getMap(data, v)
- case reflect.Slice:
- getList(data, v)
- default:
- panic(newUnmarshalError(pnType, v))
- }
- }
- err := dataError("unmarshaling", data)
- if err != nil {
- panic(err)
- }
- return
-}
-
-func rewindUnmarshal(v interface{}, data *C.pn_data_t) {
- C.pn_data_rewind(data)
- C.pn_data_next(data)
- unmarshal(v, data)
-}
-
-// Getting into an interface is driven completely by the AMQP type, since the interface{}
-// target is type-neutral.
-func getInterface(data *C.pn_data_t, v *interface{}) {
- pnType := C.pn_data_type(data)
- switch pnType {
- case C.PN_NULL, C.PN_INVALID: // No data.
- *v = nil
- case C.PN_BOOL:
- *v = bool(C.pn_data_get_bool(data))
- case C.PN_UBYTE:
- *v = uint8(C.pn_data_get_ubyte(data))
- case C.PN_BYTE:
- *v = int8(C.pn_data_get_byte(data))
- case C.PN_USHORT:
- *v = uint16(C.pn_data_get_ushort(data))
- case C.PN_SHORT:
- *v = int16(C.pn_data_get_short(data))
- case C.PN_UINT:
- *v = uint32(C.pn_data_get_uint(data))
- case C.PN_INT:
- *v = int32(C.pn_data_get_int(data))
- case C.PN_CHAR:
- *v = uint8(C.pn_data_get_char(data))
- case C.PN_ULONG:
- *v = uint64(C.pn_data_get_ulong(data))
- case C.PN_LONG:
- *v = int64(C.pn_data_get_long(data))
- case C.PN_FLOAT:
- *v = float32(C.pn_data_get_float(data))
- case C.PN_DOUBLE:
- *v = float64(C.pn_data_get_double(data))
- case C.PN_BINARY:
- *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 = Symbol(goString(C.pn_data_get_symbol(data)))
- case C.PN_MAP:
- m := make(Map)
- unmarshal(&m, data)
- *v = m
- case C.PN_LIST:
- l := make(List, 0)
- unmarshal(&l, data)
- *v = l
- default:
- panic(newUnmarshalError(pnType, v))
- }
-}
-
-// get into map pointed at by v
-func getMap(data *C.pn_data_t, v interface{}) {
- mapValue := reflect.ValueOf(v).Elem()
- mapValue.Set(reflect.MakeMap(mapValue.Type())) // Clear the map
- switch pnType := C.pn_data_type(data); pnType {
- case C.PN_MAP:
- count := int(C.pn_data_get_map(data))
- if bool(C.pn_data_enter(data)) {
- defer C.pn_data_exit(data)
- for i := 0; i < count/2; i++ {
- if bool(C.pn_data_next(data)) {
- key := reflect.New(mapValue.Type().Key())
- unmarshal(key.Interface(), data)
- if bool(C.pn_data_next(data)) {
- val := reflect.New(mapValue.Type().Elem())
- unmarshal(val.Interface(), data)
- mapValue.SetMapIndex(key.Elem(), val.Elem())
- }
- }
- }
- }
- case C.PN_INVALID: // Leave the map empty
- default:
- panic(newUnmarshalError(pnType, v))
- }
-}
-
-func getList(data *C.pn_data_t, v interface{}) {
- pnType := C.pn_data_type(data)
- if pnType != C.PN_LIST {
- panic(newUnmarshalError(pnType, v))
- }
- count := int(C.pn_data_get_list(data))
- listValue := reflect.MakeSlice(reflect.TypeOf(v).Elem(), count, count)
- if bool(C.pn_data_enter(data)) {
- for i := 0; i < count; i++ {
- if bool(C.pn_data_next(data)) {
- val := reflect.New(listValue.Type().Elem())
- unmarshal(val.Interface(), data)
- listValue.Index(i).Set(val.Elem())
- }
- }
- C.pn_data_exit(data)
- }
- reflect.ValueOf(v).Elem().Set(listValue)
-}
-
-// decode from bytes.
-// Return bytes decoded or 0 if we could not decode a complete object.
-//
-func decode(data *C.pn_data_t, bytes []byte) int {
- if len(bytes) == 0 {
- return 0
- }
- 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(internal.Errorf("unmarshal %s", internal.PnErrorCode(n)))
- }
- return n
-}
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/478ba4ea/proton-c/bindings/go/src/qpid.apache.org/proton/amqp/url.go
----------------------------------------------------------------------
diff --git a/proton-c/bindings/go/src/qpid.apache.org/proton/amqp/url.go b/proton-c/bindings/go/src/qpid.apache.org/proton/amqp/url.go
deleted file mode 100644
index 7a4ef13..0000000
--- a/proton-c/bindings/go/src/qpid.apache.org/proton/amqp/url.go
+++ /dev/null
@@ -1,96 +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 amqp
-
-/*
-#include <stdlib.h>
-#include <string.h>
-#include <proton/url.h>
-
-// Helper function for setting URL fields.
-typedef void (*setter_fn)(pn_url_t* url, const char* value);
-inline void set(pn_url_t *url, setter_fn s, const char* value) {
- s(url, value);
-}
-*/
-import "C"
-
-import (
- "net"
- "net/url"
- "qpid.apache.org/proton/internal"
- "unsafe"
-)
-
-const (
- amqp string = "amqp"
- amqps = "amqps"
-)
-
-// ParseUrl parses an AMQP URL string and returns a net/url.Url.
-//
-// It is more forgiving than net/url.Parse and allows most of the parts of the
-// URL to be missing, assuming AMQP defaults.
-//
-func ParseURL(s string) (u *url.URL, err error) {
- cstr := C.CString(s)
- defer C.free(unsafe.Pointer(cstr))
- pnUrl := C.pn_url_parse(cstr)
- if pnUrl == nil {
- return nil, internal.Errorf("bad URL %#v", s)
- }
- defer C.pn_url_free(pnUrl)
-
- scheme := C.GoString(C.pn_url_get_scheme(pnUrl))
- username := C.GoString(C.pn_url_get_username(pnUrl))
- password := C.GoString(C.pn_url_get_password(pnUrl))
- host := C.GoString(C.pn_url_get_host(pnUrl))
- port := C.GoString(C.pn_url_get_port(pnUrl))
- path := C.GoString(C.pn_url_get_path(pnUrl))
-
- if err != nil {
- return nil, internal.Errorf("bad URL %#v: %s", s, err)
- }
- if scheme == "" {
- scheme = amqp
- }
- if port == "" {
- if scheme == amqps {
- port = amqps
- } else {
- port = amqp
- }
- }
- var user *url.Userinfo
- if password != "" {
- user = url.UserPassword(username, password)
- } else if username != "" {
- user = url.User(username)
- }
-
- u = &url.URL{
- Scheme: scheme,
- User: user,
- Host: net.JoinHostPort(host, port),
- Path: path,
- }
-
- return u, nil
-}
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/478ba4ea/proton-c/bindings/go/src/qpid.apache.org/proton/amqp/url_test.go
----------------------------------------------------------------------
diff --git a/proton-c/bindings/go/src/qpid.apache.org/proton/amqp/url_test.go b/proton-c/bindings/go/src/qpid.apache.org/proton/amqp/url_test.go
deleted file mode 100644
index f80f1c4..0000000
--- a/proton-c/bindings/go/src/qpid.apache.org/proton/amqp/url_test.go
+++ /dev/null
@@ -1,51 +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 amqp
-
-import (
- "fmt"
-)
-
-func ExampleParseURL() {
- for _, s := range []string{
- "amqp://username:password@host:1234/path",
- "host:1234",
- "host",
- ":1234",
- "host/path",
- "amqps://host",
- "",
- } {
- u, err := ParseURL(s)
- if err != nil {
- fmt.Println(err)
- } else {
- fmt.Println(u)
- }
- }
- // Output:
- // amqp://username:password@host:1234/path
- // amqp://host:1234
- // amqp://host:amqp
- // amqp://:1234
- // amqp://host:amqp/path
- // amqps://host:amqps
- // proton: bad URL ""
-}
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/478ba4ea/proton-c/bindings/go/src/qpid.apache.org/proton/concurrent/connection.go
----------------------------------------------------------------------
diff --git a/proton-c/bindings/go/src/qpid.apache.org/proton/concurrent/connection.go b/proton-c/bindings/go/src/qpid.apache.org/proton/concurrent/connection.go
deleted file mode 100644
index 63fd3fc..0000000
--- a/proton-c/bindings/go/src/qpid.apache.org/proton/concurrent/connection.go
+++ /dev/null
@@ -1,213 +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 concurrent
-
-// #include <proton/disposition.h>
-import "C"
-
-import (
- "net"
- "qpid.apache.org/proton"
- "qpid.apache.org/proton/internal"
- "sync"
-)
-
-// Connection is an AMQP connection, created by a Container.
-type Connection interface {
- Endpoint
-
- // Sender opens a new sender on the DefaultSession.
- //
- // v can be a string, which is used as the Target address, or a SenderSettings
- // struct containing more details settings.
- Sender(v interface{}) (Sender, error)
-
- // Receiver opens a new Receiver on the DefaultSession().
- //
- // v can be a string, which is used as the
- // Source address, or a ReceiverSettings struct containing more details
- // settings.
- Receiver(v interface{}) (Receiver, error)
-
- // Server puts the connection in server mode, must be called before Open().
- //
- // A server connection will do protocol negotiation to accept a incoming AMQP
- // connection. Normally you would call this for a connection accepted by
- // net.Listener.Accept()
- //
- Server()
-
- // Listen enables endpoints opened by the remote peer to be accepted by calling Accept().
- // Must be called before Open().
- Listen()
-
- // DefaultSession() returns a default session for the connection. It is opened
- // on the first call to DefaultSession and returned on subsequent calls.
- DefaultSession() (Session, error)
-
- // Session opens a new session.
- Session() (Session, error)
-
- // Accept returns the next Endpoint (Session, Sender or Receiver) opened by
- // the remote peer. It returns (nil, error) if the connection closes.
- //
- // You must call Endpoint.Open() to complete opening the returned Endpoint or
- // Endpoint.Close(error) to reject it. You can set endpoint properties before
- // calling Open()
- //
- // You can use a type switch or type conversion to test which kind of Endpoint
- // has been returned.
- //
- // You must call Connection.Listen() before Connection.Open() to enable Accept.
- //
- // The connection buffers endpoints until you call Accept() so normally you
- // should have a dedicated goroutine calling Accept() in a loop to process it
- // rapidly.
- //
- Accept() (Endpoint, error)
-
- // Container for the connection.
- Container() Container
-
- // Disconnect the connection abruptly with an error.
- Disconnect(error)
-}
-
-type connection struct {
- endpoint
- listenOnce, defaultSessionOnce sync.Once
-
- // Set before Open()
- container *container
- conn net.Conn
- incoming *internal.FlexChannel
-
- // Set by Open()
- handler *handler
- engine *proton.Engine
- err internal.FirstError
- eConnection proton.Connection
-
- defaultSession Session
-}
-
-func newConnection(conn net.Conn, cont *container) (*connection, error) {
- c := &connection{container: cont, conn: conn}
- c.handler = newHandler(c)
- var err error
- c.engine, err = proton.NewEngine(c.conn, c.handler.delegator)
- if err != nil {
- return nil, err
- }
- c.str = c.engine.String()
- c.eConnection = c.engine.Connection()
- return c, nil
-}
-
-func (c *connection) Server() { c.engine.Server() }
-
-func (c *connection) Listen() {
- c.listenOnce.Do(func() { c.incoming = internal.NewFlexChannel(-1) })
-}
-
-func (c *connection) Open() error {
- go c.engine.Run()
- return nil
-}
-
-func (c *connection) Close(err error) {
- c.engine.Close(err)
- c.setError(c.engine.Error()) // Will be io.EOF on close OK
- if c.incoming != nil {
- close(c.incoming.In)
- }
-}
-
-func (c *connection) Disconnect(err error) {
- c.engine.Disconnect(err)
- if c.incoming != nil {
- close(c.incoming.In)
- }
-}
-
-func (c *connection) closed(err error) {
- // Call from another goroutine to initiate close without deadlock.
- go c.Close(err)
-}
-
-func (c *connection) Session() (Session, error) {
- var s Session
- err := c.engine.InjectWait(func() error {
- eSession, err := c.engine.Connection().Session()
- if err == nil {
- eSession.Open()
- if err == nil {
- s = newSession(c, eSession)
- }
- }
- return err
- })
- return s, err
-}
-
-func (c *connection) handleIncoming(sn *session, l proton.Link) {
- if l.IsReceiver() {
- c.incoming.In <- newReceiver(makeIncomingLink(sn, l))
- } else {
- c.incoming.In <- newSender(makeIncomingLink(sn, l))
- }
-}
-
-func (c *connection) Accept() (Endpoint, error) {
- v, ok := <-c.incoming.Out
- if !ok {
- return nil, c.Error()
- } else {
- return v.(Endpoint), nil
- }
-}
-
-func (c *connection) Container() Container { return c.container }
-
-func (c *connection) DefaultSession() (s Session, err error) {
- c.defaultSessionOnce.Do(func() {
- c.defaultSession, err = c.Session()
- })
- if err == nil {
- err = c.Error()
- }
- return c.defaultSession, err
-}
-
-func (c *connection) Sender(v interface{}) (Sender, error) {
- if s, err := c.DefaultSession(); err == nil {
- return s.Sender(v)
- } else {
- return nil, err
- }
-}
-
-func (c *connection) Receiver(v interface{}) (Receiver, error) {
- if s, err := c.DefaultSession(); err == nil {
- return s.Receiver(v)
- } else {
- return nil, err
- }
-}
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/478ba4ea/proton-c/bindings/go/src/qpid.apache.org/proton/concurrent/container.go
----------------------------------------------------------------------
diff --git a/proton-c/bindings/go/src/qpid.apache.org/proton/concurrent/container.go b/proton-c/bindings/go/src/qpid.apache.org/proton/concurrent/container.go
deleted file mode 100644
index 5edecfc..0000000
--- a/proton-c/bindings/go/src/qpid.apache.org/proton/concurrent/container.go
+++ /dev/null
@@ -1,71 +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 concurrent
-
-import (
- "net"
- "qpid.apache.org/proton/internal"
-)
-
-// Container is an AMQP container, it represents a single AMQP "application".It
-// provides functions to create new Connections to remote containers.
-//
-// Create with NewContainer()
-//
-type Container interface {
- // Id is a unique identifier for the container in your distributed application.
- Id() string
-
- // Create a new AMQP Connection over the supplied net.Conn connection.
- //
- // You must call Connection.Open() on the returned Connection, after
- // setting any Connection properties you need to set. Note the net.Conn
- // can be an outgoing connection (e.g. made with net.Dial) or an incoming
- // connection (e.g. made with net.Listener.Accept())
- Connection(conn net.Conn) (Connection, error)
-}
-
-type container struct {
- id string
- linkNames internal.IdCounter
-}
-
-// NewContainer creates a new container. The id must be unique in your
-// distributed application, all connections created by the container
-// will have this container-id.
-//
-// If id == "" a random UUID will be generated for the id.
-func NewContainer(id string) Container {
- if id == "" {
- id = internal.UUID4().String()
- }
- cont := &container{id: id}
- return cont
-}
-
-func (cont *container) Id() string { return cont.id }
-
-func (cont *container) nextLinkName() string {
- return cont.id + "@" + cont.linkNames.Next()
-}
-
-func (cont *container) Connection(conn net.Conn) (Connection, error) {
- return newConnection(conn, cont)
-}
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/478ba4ea/proton-c/bindings/go/src/qpid.apache.org/proton/concurrent/doc.go
----------------------------------------------------------------------
diff --git a/proton-c/bindings/go/src/qpid.apache.org/proton/concurrent/doc.go b/proton-c/bindings/go/src/qpid.apache.org/proton/concurrent/doc.go
deleted file mode 100644
index 810a5da..0000000
--- a/proton-c/bindings/go/src/qpid.apache.org/proton/concurrent/doc.go
+++ /dev/null
@@ -1,46 +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 concurrent provides a procedural, concurrent Go API for exchanging AMQP
-messages. You can write clients or servers using this API.
-
-Start by creating a Container with NewContainer. A Container represents a client
-or server application that can contain incoming or outgoing connections.
-
-You can create connections with the standard Go 'net' package using net.Dial or
-net.Listen. Create an AMQP connection over a net.Conn with
-Container.Connection() and open it with Connection.Open().
-
-AMQP sends messages over "links", each link has a Sender and Receiver
-end. Connection.Sender() and Connection.Receiver() allow you to create links to
-send and receive messages.
-
-You can also create an AMQP server connection by calling Connection.Listen()
-before calling Open() on the connection. You can then call Connection.Accept()
-after calling Connection.Open() to accept incoming sessions and links.
-
-*/
-package concurrent
-
-//#cgo LDFLAGS: -lqpid-proton
-import "C"
-
-// Just for package comment
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/478ba4ea/proton-c/bindings/go/src/qpid.apache.org/proton/concurrent/endpoint.go
----------------------------------------------------------------------
diff --git a/proton-c/bindings/go/src/qpid.apache.org/proton/concurrent/endpoint.go b/proton-c/bindings/go/src/qpid.apache.org/proton/concurrent/endpoint.go
deleted file mode 100644
index f647058..0000000
--- a/proton-c/bindings/go/src/qpid.apache.org/proton/concurrent/endpoint.go
+++ /dev/null
@@ -1,87 +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 concurrent
-
-import (
- "io"
- "qpid.apache.org/proton"
- "qpid.apache.org/proton/internal"
-)
-
-// Closed is an alias for io.EOF. It is returned as an error when an endpoint
-// was closed cleanly.
-var Closed = io.EOF
-
-// Endpoint is the common interface for Connection, Session, Link, Sender and Receiver.
-//
-// Endpoints can be created locally or by the remote peer. You must Open() an
-// endpoint before you can use it. Some endpoints have additional Set*() methods
-// that must be called before Open() to take effect, see Connection, Session,
-// Link, Sender and Receiver for details.
-//
-type Endpoint interface {
- // Open the local end of a remotely-initiated endpoint. You must Open()
- // endpoints returned by Connection.Accept() before using them.
- Open() error
-
- // Close an endpoint and signal an error to the remote end if error != nil.
- Close(error)
-
- // String is a human readable identifier, useful for debugging and logging.
- String() string
-
- // Error returns nil if the endpoint is open, otherwise returns an error.
- // Error() == Closed means the endpoint was closed without error.
- Error() error
-}
-
-// Implements setError() and Error() from Endpoint values that hold an error.
-type errorHolder struct {
- err internal.FirstError
-}
-
-func (e *errorHolder) setError(err error) error { return e.err.Set(err) }
-func (e *errorHolder) Error() error { return e.err.Get() }
-
-// Implements Error() and String() from Endpoint
-type endpoint struct {
- errorHolder
- str string // Must be set by the value that embeds endpoint.
-}
-
-func (e *endpoint) String() string { return e.str }
-
-// Call in proton goroutine to close an endpoint locally
-// handler will complete the close when remote end closes.
-func localClose(ep proton.Endpoint, err error) {
- if ep.State().LocalActive() {
- if err != nil {
- ep.Condition().SetError(err)
- }
- ep.Close()
- }
-}
-
-func (e *endpoint) closeError(err error) {
- if err == nil {
- err = Closed
- }
- e.err.Set(err)
-}
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/478ba4ea/proton-c/bindings/go/src/qpid.apache.org/proton/concurrent/handler.go
----------------------------------------------------------------------
diff --git a/proton-c/bindings/go/src/qpid.apache.org/proton/concurrent/handler.go b/proton-c/bindings/go/src/qpid.apache.org/proton/concurrent/handler.go
deleted file mode 100644
index bf8fcd3..0000000
--- a/proton-c/bindings/go/src/qpid.apache.org/proton/concurrent/handler.go
+++ /dev/null
@@ -1,137 +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 concurrent
-
-import (
- "qpid.apache.org/proton"
- "qpid.apache.org/proton/amqp"
-)
-
-// NOTE: methods in this file are called only in the proton goroutine unless otherwise indicated.
-
-type handler struct {
- delegator *proton.MessagingDelegator
- connection *connection
- links map[proton.Link]Link
- sentMessages map[proton.Delivery]*sentMessage
- sessions map[proton.Session]*session
-}
-
-func newHandler(c *connection) *handler {
- h := &handler{
- connection: c,
- links: make(map[proton.Link]Link),
- sentMessages: make(map[proton.Delivery]*sentMessage),
- sessions: make(map[proton.Session]*session),
- }
- h.delegator = proton.NewMessagingDelegator(h)
- // Disable auto features of MessagingDelegator, we do these ourselves.
- h.delegator.Prefetch = 0
- h.delegator.AutoAccept = false
- h.delegator.AutoSettle = false
- h.delegator.AutoOpen = false
- return h
-}
-
-func (h *handler) HandleMessagingEvent(t proton.MessagingEvent, e proton.Event) {
- switch t {
-
- case proton.MMessage:
- if r, ok := h.links[e.Link()].(*receiver); ok {
- r.handleDelivery(e.Delivery())
- } else {
- h.connection.closed(
- amqp.Errorf(amqp.InternalError, "cannot find receiver for link %s", e.Link()))
- }
-
- case proton.MSettled:
- if sm := h.sentMessages[e.Delivery()]; sm != nil {
- sm.settled(nil)
- }
-
- case proton.MSessionOpening:
- if e.Session().State().LocalUninit() { // Remotely opened
- s := newSession(h.connection, e.Session())
- h.sessions[e.Session()] = s
- if h.connection.incoming != nil {
- h.connection.incoming.In <- s
- } else {
- proton.CloseError(e.Session(), amqp.Errorf(amqp.NotAllowed, "remote sessions not allowed"))
- }
- }
-
- case proton.MSessionClosing:
- e.Session().Close()
-
- case proton.MSessionClosed:
- err := e.Session().RemoteCondition().Error()
- for l, _ := range h.links {
- if l.Session() == e.Session() {
- h.linkClosed(l, err)
- }
- }
- delete(h.sessions, e.Session())
-
- case proton.MLinkOpening:
- l := e.Link()
- if l.State().LocalUninit() { // Remotely opened
- if h.connection.incoming == nil {
- proton.CloseError(l, amqp.Errorf(amqp.NotAllowed, ("no remote links")))
- break
- }
- s := h.sessions[l.Session()]
- if s == nil {
- proton.CloseError(
- l, amqp.Errorf(amqp.InternalError, ("cannot find session for link")))
- break
- }
- h.connection.handleIncoming(s, l)
- }
-
- case proton.MLinkClosing:
- e.Link().Close()
-
- case proton.MLinkClosed:
- h.linkClosed(e.Link(), e.Link().RemoteCondition().Error())
-
- case proton.MDisconnected:
- err := h.connection.Error()
- for l, _ := range h.links {
- h.linkClosed(l, err)
- }
- for _, s := range h.sessions {
- s.closed(err)
- }
- for _, sm := range h.sentMessages {
- sm.settled(err)
- }
- }
-}
-
-func (h *handler) linkClosed(l proton.Link, err error) {
- if link := h.links[l]; link != nil {
- link.closed(err)
- delete(h.links, l)
- }
-}
-
-func (h *handler) addLink(rl proton.Link, ll Link) {
- h.links[rl] = ll
-}
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/478ba4ea/proton-c/bindings/go/src/qpid.apache.org/proton/concurrent/link.go
----------------------------------------------------------------------
diff --git a/proton-c/bindings/go/src/qpid.apache.org/proton/concurrent/link.go b/proton-c/bindings/go/src/qpid.apache.org/proton/concurrent/link.go
deleted file mode 100644
index cf4b8aa..0000000
--- a/proton-c/bindings/go/src/qpid.apache.org/proton/concurrent/link.go
+++ /dev/null
@@ -1,232 +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 concurrent
-
-import (
- "qpid.apache.org/proton"
- "qpid.apache.org/proton/internal"
-)
-
-type LinkSettings struct {
- // Source address that messages come from.
- Source string
- // Target address that messages are going to.
- Target string
-
- // Unique (per container) name of the link.
- // Leave blank to have the container generate a unique name automatically.
- Name string
-
- // SndSettleMode defines when the sending end of the link settles message delivery.
- // Can set via AtMostOnce or AtLeastOnce.
- SndSettleMode SndSettleMode
-
- // RcvSettleMode defines when the receiving end of the link settles message delivery.
- RcvSettleMode RcvSettleMode
-}
-
-// AtMostOnce sets "fire and forget" mode, messages are sent but no
-// acknowledgment is received, messages can be lost if there is a network
-// failure. Sets SndSettleMode=SendSettled and RcvSettleMode=RcvFirst
-func (s *LinkSettings) AtMostOnce() {
- s.SndSettleMode = SndSettled
- s.RcvSettleMode = RcvFirst
-}
-
-// AtLeastOnce requests acknowledgment for every message, acknowledgment
-// indicates the message was definitely received. In the event of a
-// failure, unacknowledged messages can be re-sent but there is a chance
-// that the message will be received twice in this case.
-// Sets SndSettleMode=SndUnsettled and RcvSettleMode=RcvFirst
-func (s *LinkSettings) AtLeastOnce() {
- s.SndSettleMode = SndUnsettled
- s.RcvSettleMode = RcvFirst
-}
-
-// Link is the common interface for Sender and Receiver links.
-type Link interface {
- Endpoint
-
- // Settings for this link.
- Settings() LinkSettings
-
- IsSender() bool
- IsReceiver() bool
-
- IsOpen() bool
-
- // Credit indicates how many messages the receiving end of the link can accept.
- //
- // A Receiver adds credit automatically when it can accept more messages.
- //
- // On a Sender credit can be negative, meaning that messages in excess of the
- // receiver's credit limit have been buffered locally till credit is available.
- Credit() (int, error)
-
- // Called in event loop on closed event.
- closed(err error)
-}
-
-// SndSettleMode defines when the sending end of the link settles message delivery.
-// Can set via AtMostOnce or AtLeastOnce.
-type SndSettleMode proton.SndSettleMode
-
-const (
- // Messages are sent unsettled
- SndUnsettled = SndSettleMode(proton.SndUnsettled)
- // Messages are sent already settled
- SndSettled = SndSettleMode(proton.SndSettled)
- // Sender can send either unsettled or settled messages.
- SendMixed = SndSettleMode(proton.SndMixed)
-)
-
-// RcvSettleMode defines when the receiving end of the link settles message delivery.
-type RcvSettleMode proton.RcvSettleMode
-
-const (
- // Receiver settles first.
- RcvFirst = RcvSettleMode(proton.RcvFirst)
- // Receiver waits for sender to settle before settling.
- RcvSecond = RcvSettleMode(proton.RcvSecond)
-)
-
-// Implement Link interface, for embedding in sender and receiver.
-//
-// Link creation: there are two ways to create a link.
-//
-// Session.NewSender() and Session.NewReceiver() create a "local" link which has
-// the session and isSender fields set. The user can set properties like Name,
-// Target and Source. On Open() the eLink is created and the properties are set
-// on the eLink.
-//
-// An "incoming" is created by the connection. with session, isSender, name,
-// source, target all set from the incoming eLink, these properties cannot be
-// changed by the user. There may be other properties (e.g. Receiver.SetCapacity())
-// that can be set by the user before Open().
-//
-type link struct {
- endpoint
-
- settings LinkSettings
- session *session
- eLink proton.Link
- isOpen bool
- isSender bool
-}
-
-func (l *link) Settings() LinkSettings { return l.settings }
-func (l *link) IsSender() bool { return l.isSender }
-func (l *link) IsReceiver() bool { return !l.isSender }
-func (l *link) IsOpen() bool { return l.isOpen }
-func (l *link) Session() Session { return l.session }
-func (l *link) Connection() Connection { return l.session.Connection() }
-
-func (l *link) engine() *proton.Engine { return l.session.connection.engine }
-func (l *link) handler() *handler { return l.session.connection.handler }
-
-// initLocal initializes a local link associated with a session.
-// Call in proton goroutine
-func makeLocalLink(sn *session, isSender bool, settings LinkSettings) (link, error) {
- var l link
- l.session = sn
- l.settings = settings
- l.isSender = isSender
- if l.settings.Name == "" {
- l.settings.Name = l.session.connection.container.nextLinkName()
- }
- if l.IsSender() {
- l.eLink = l.session.eSession.Sender(l.settings.Name)
- } else {
- l.eLink = l.session.eSession.Receiver(l.settings.Name)
- }
- if l.eLink.IsNil() {
- return l, l.setError(internal.Errorf("cannot create link %s", l))
- }
- l.setSettings()
- return l, nil
-}
-
-// Set local end of the link to match LinkSettings.
-func (l *link) setSettings() {
- l.eLink.Source().SetAddress(l.settings.Source)
- l.eLink.Target().SetAddress(l.settings.Target)
- l.eLink.SetSndSettleMode(proton.SndSettleMode(l.settings.SndSettleMode))
- l.eLink.SetRcvSettleMode(proton.RcvSettleMode(l.settings.RcvSettleMode))
- l.str = l.eLink.String()
-}
-
-// initIncoming sets link settings from an incoming proton.Link.
-// Call in proton goroutine
-func makeIncomingLink(sn *session, eLink proton.Link) link {
- l := link{
- session: sn,
- isSender: eLink.IsSender(),
- eLink: eLink,
- settings: LinkSettings{
- Source: eLink.RemoteSource().Address(),
- Target: eLink.RemoteTarget().Address(),
- Name: eLink.Name(),
- SndSettleMode: SndSettleMode(eLink.RemoteSndSettleMode()),
- RcvSettleMode: RcvSettleMode(eLink.RemoteRcvSettleMode()),
- },
- }
- l.setSettings()
- return l
-}
-
-func (l *link) setPanicIfOpen() {
- if l.IsOpen() {
- panic(internal.Errorf("link is already open %s", l))
- }
-}
-
-// open a link, local or incoming. Call in proton goroutine
-func (l *link) open() error {
- if l.Error() != nil {
- return l.Error()
- }
- l.eLink.Open()
- l.isOpen = true
- return nil
-}
-
-// Called in proton goroutine
-func (l *link) closed(err error) {
- l.setError(err)
- if l.eLink.State().RemoteActive() {
- if l.Error() != nil {
- l.eLink.Condition().SetError(l.Error())
- }
- l.eLink.Close()
- }
- l.setError(Closed) // If no error set, mark as closed.
-}
-
-func (l *link) Credit() (credit int, err error) {
- err = l.engine().InjectWait(func() error {
- credit = l.eLink.Credit()
- return nil
- })
- return
-}
-
-func (l *link) Close(err error) {
- l.engine().Inject(func() { localClose(l.eLink, err) })
-}
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/478ba4ea/proton-c/bindings/go/src/qpid.apache.org/proton/concurrent/messaging_test.go
----------------------------------------------------------------------
diff --git a/proton-c/bindings/go/src/qpid.apache.org/proton/concurrent/messaging_test.go b/proton-c/bindings/go/src/qpid.apache.org/proton/concurrent/messaging_test.go
deleted file mode 100644
index 0ee9f1a..0000000
--- a/proton-c/bindings/go/src/qpid.apache.org/proton/concurrent/messaging_test.go
+++ /dev/null
@@ -1,205 +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 concurrent
-
-import (
- "fmt"
- "net"
- "qpid.apache.org/proton/amqp"
- "testing"
- "time"
-)
-
-func panicIf(err error) {
- if err != nil {
- panic(err)
- }
-}
-
-// Start a server, return listening addr and channel for incoming Connection.
-func newServer(cont Container) (net.Addr, <-chan Connection) {
- listener, err := net.Listen("tcp", "")
- panicIf(err)
- addr := listener.Addr()
- ch := make(chan Connection)
- go func() {
- conn, err := listener.Accept()
- c, err := cont.Connection(conn)
- panicIf(err)
- c.Server()
- c.Listen()
- panicIf(c.Open())
- ch <- c
- }()
- return addr, ch
-}
-
-// Return open an client connection and session, return the session.
-func newClient(cont Container, addr net.Addr) Session {
- conn, err := net.Dial(addr.Network(), addr.String())
- panicIf(err)
- c, err := cont.Connection(conn)
- panicIf(err)
- c.Open()
- sn, err := c.Session()
- panicIf(err)
- panicIf(sn.Open())
- return sn
-}
-
-// Return client and server ends of the same connection.
-func newClientServer() (client Session, server Connection) {
- addr, ch := newServer(NewContainer(""))
- client = newClient(NewContainer(""), addr)
- return client, <-ch
-}
-
-// Close client and server
-func closeClientServer(client Session, server Connection) {
- client.Connection().Close(nil)
- server.Close(nil)
-}
-
-// Send a message one way with a client sender and server receiver, verify ack.
-func TestClientSendServerReceive(t *testing.T) {
- var err error
- client, server := newClientServer()
- defer func() {
- closeClientServer(client, server)
- }()
-
- timeout := time.Second * 2
- nLinks := 3
- nMessages := 3
-
- s := make([]Sender, nLinks)
- for i := 0; i < nLinks; i++ {
- s[i], err = client.Sender(fmt.Sprintf("foo%d", i))
- if err != nil {
- t.Fatal(err)
- }
- }
-
- // Server accept session and receivers
- ep, err := server.Accept()
- ep.Open() // Accept incoming session
- r := make([]Receiver, nLinks)
- for i := 0; i < nLinks; i++ { // Accept incoming receivers
- ep, err = server.Accept()
- r[i] = ep.(Receiver)
- err = r[i].Open()
- if err != nil {
- t.Fatal(err)
- }
- }
-
- for i := 0; i < nLinks; i++ {
- for j := 0; j < nMessages; j++ {
- // Client send
- m := amqp.NewMessageWith(fmt.Sprintf("foobar%v-%v", i, j))
- sm, err := s[i].Send(m)
- if err != nil {
- t.Fatal(err)
- }
-
- // Server recieve
- rm, err := r[i].Receive()
- if err != nil {
- t.Fatal(err)
- }
- if want, got := interface{}(fmt.Sprintf("foobar%v-%v", i, j)), rm.Message.Body(); want != got {
- t.Errorf("%#v != %#v", want, got)
- }
-
- // Should not be acknowledged on client yet
- if d, err := sm.DispositionTimeout(0); err != Timeout || NoDisposition != d {
- t.Errorf("want [no-disposition/timeout] got [%v/%v]", d, err)
- }
- // Server ack
- if err := rm.Acknowledge(Rejected); err != nil {
- t.Error(err)
- }
- // Client get ack.
- if d, err := sm.DispositionTimeout(timeout); err != nil || Rejected != d {
- t.Errorf("want [rejected/nil] got [%v/%v]", d, err)
- }
- }
- }
-}
-
-func TestClientReceiver(t *testing.T) {
- client, server := newClientServer()
- nMessages := 3
-
- done := make(chan struct{})
- go func() { // Server sends
- defer close(done)
- for {
- ep, err := server.Accept()
- switch {
- case err == Closed:
- return
- case err == nil:
- break
- default:
- t.Error(err)
- return
- }
- ep.Open()
- if s, ok := ep.(Sender); ok {
- go func() {
- for i := int32(0); i < int32(nMessages); i++ {
- sm, err := s.Send(amqp.NewMessageWith(i))
- if err != nil {
- t.Error(err)
- return
- } else {
- sm.Disposition() // Sync send.
- }
- }
- s.Close(nil)
- }()
- }
- }
- }()
-
- r, err := client.Receiver("foo")
- if err != nil {
- t.Fatal(err)
- }
- for i := int32(0); i < int32(nMessages); i++ {
- rm, err := r.Receive()
- if err != nil {
- if err != Closed {
- t.Error(err)
- }
- break
- }
- if err := rm.Accept(); err != nil {
- t.Error(err)
- }
- if b, ok := rm.Message.Body().(int32); !ok || b != i {
- t.Errorf("want %v, true got %v, %v", i, b, ok)
- }
- }
- server.Close(nil)
- <-done
- client.Connection().Close(nil)
-}
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/478ba4ea/proton-c/bindings/go/src/qpid.apache.org/proton/concurrent/receiver.go
----------------------------------------------------------------------
diff --git a/proton-c/bindings/go/src/qpid.apache.org/proton/concurrent/receiver.go b/proton-c/bindings/go/src/qpid.apache.org/proton/concurrent/receiver.go
deleted file mode 100644
index 5bcf9f2..0000000
--- a/proton-c/bindings/go/src/qpid.apache.org/proton/concurrent/receiver.go
+++ /dev/null
@@ -1,241 +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 concurrent
-
-import (
- "qpid.apache.org/proton"
- "qpid.apache.org/proton/amqp"
- "qpid.apache.org/proton/internal"
- "time"
-)
-
-type ReceiverSettings struct {
- LinkSettings
-
- // Capacity is the number of messages that the receiver can buffer locally.
- // If unset (zero) it will be set to 1.
- Capacity int
-
- // Prefetch==true means the Receiver will automatically issue credit to the
- // remote sender to keep its buffer as full as possible, i.e. it will
- // "pre-fetch" messages independently of the application calling
- // Receive(). This gives good throughput for applications that handle a
- // continuous stream of messages. Larger capacity may improve throughput, the
- // optimal value depends on the characteristics of your application.
- //
- // Prefetch==false means the Receiver will issue only issue credit when you
- // call Receive(), and will only issue enough credit to satisfy the calls
- // actually made. This gives lower throughput but will not fetch any messages
- // in advance. It is good for synchronous applications that need to evaluate
- // each message before deciding whether to receive another. The
- // request-response pattern is a typical example. If you make concurrent
- // calls to Receive with pre-fetch disabled, you can improve performance by
- // setting the capacity close to the expected number of concurrent calls.
- //
- Prefetch bool
-}
-
-// Receiver is a Link that receives messages.
-//
-type Receiver interface {
- Link
-
- // SetCapacity sets the Capacity and Prefetch (see ReceiverSettings) It may
- // may called before Open() on an accepted receiver, it cannot be changed once
- // the receiver is Open().
- SetCapacity(capacity int, prefetch bool)
-
- // Receive blocks until a message is available or until the Receiver is closed
- // and has no more buffered messages.
- Receive() (ReceivedMessage, error)
-
- // ReceiveTimeout is like Receive but gives up after timeout, see Timeout.
- ReceiveTimeout(timeout time.Duration) (ReceivedMessage, error)
-}
-
-// Flow control policy for a receiver.
-type policy interface {
- // Called at the start of Receive() to adjust credit before fetching a message.
- Pre(*receiver)
- // Called after Receive() has received a message from Buffer() before it returns.
- // Non-nil error means no message was received because of an error.
- Post(*receiver, error)
-}
-
-type prefetchPolicy struct{}
-
-func (p prefetchPolicy) Flow(r *receiver) {
- r.engine().Inject(func() {
- _, _, max := r.credit()
- if max > 0 {
- r.eLink.Flow(max)
- }
- })
-}
-func (p prefetchPolicy) Pre(r *receiver) { p.Flow(r) }
-func (p prefetchPolicy) Post(r *receiver, err error) {
- if err == nil {
- p.Flow(r)
- }
-}
-
-type noPrefetchPolicy struct{ waiting int }
-
-func (p noPrefetchPolicy) Flow(r *receiver) { // Not called in proton goroutine
- r.engine().Inject(func() {
- len, credit, max := r.credit()
- add := p.waiting - (len + credit)
- if add > max {
- add = max // Don't overflow
- }
- if add > 0 {
- r.eLink.Flow(add)
- }
- })
-}
-func (p noPrefetchPolicy) Pre(r *receiver) { p.waiting++; p.Flow(r) }
-func (p noPrefetchPolicy) Post(r *receiver, err error) {
- p.waiting--
- if err == nil {
- p.Flow(r)
- }
-}
-
-// Receiver implementation
-type receiver struct {
- link
- // Set in Setup()
- capacity int
- prefetch bool
-
- // Set in Open()
- buffer chan ReceivedMessage
- policy policy
-}
-
-func newReceiver(l link) Receiver { return &receiver{link: l} }
-
-func (r *receiver) SetCapacity(capacity int, prefetch bool) {
- r.setPanicIfOpen()
- if capacity < 1 {
- capacity = 1
- }
- r.capacity = capacity
- r.prefetch = prefetch
-}
-
-// Accept and open an incoming receiver.
-func (r *receiver) Open() error {
- if r.capacity == 0 {
- r.SetCapacity(1, false)
- }
- if r.prefetch {
- r.policy = &prefetchPolicy{}
- } else {
- r.policy = &noPrefetchPolicy{}
- }
- err := r.engine().InjectWait(func() error {
- err := r.open()
- if err == nil {
- r.buffer = make(chan ReceivedMessage, r.capacity)
- r.handler().addLink(r.eLink, r)
- }
- return err
- })
- return r.setError(err)
-}
-
-// call in proton goroutine
-func (r *receiver) credit() (buffered, credit, capacity int) {
- return len(r.buffer), r.eLink.Credit(), cap(r.buffer)
-}
-
-func (r *receiver) Capacity() int { return cap(r.buffer) }
-
-func (r *receiver) Receive() (rm ReceivedMessage, err error) {
- return r.ReceiveTimeout(Forever)
-}
-
-func (r *receiver) ReceiveTimeout(timeout time.Duration) (rm ReceivedMessage, err error) {
- internal.Assert(r.buffer != nil, "Receiver is not open: %s", r)
- r.policy.Pre(r)
- defer func() { r.policy.Post(r, err) }()
- rmi, ok, timedout := timedReceive(r.buffer, timeout)
- switch {
- case timedout:
- return ReceivedMessage{}, Timeout
- case !ok:
- return ReceivedMessage{}, r.Error()
- default:
- return rmi.(ReceivedMessage), nil
- }
-}
-
-// Called in proton goroutine
-func (r *receiver) handleDelivery(delivery proton.Delivery) {
- if r.eLink.State().RemoteClosed() {
- localClose(r.eLink, r.eLink.RemoteCondition().Error())
- return
- }
- if delivery.HasMessage() {
- m, err := delivery.Message()
- if err != nil {
- localClose(r.eLink, err)
- return
- }
- internal.Assert(m != nil)
- r.eLink.Advance()
- if r.eLink.Credit() < 0 {
- localClose(r.eLink, internal.Errorf("received message in excess of credit limit"))
- } else {
- // We never issue more credit than cap(buffer) so this will not block.
- r.buffer <- ReceivedMessage{m, delivery, r}
- }
- }
-}
-
-func (r *receiver) closed(err error) {
- r.closeError(err)
- close(r.buffer)
-}
-
-// ReceivedMessage contains an amqp.Message and allows the message to be acknowledged.
-type ReceivedMessage struct {
- // Message is the received message.
- Message amqp.Message
-
- eDelivery proton.Delivery
- receiver Receiver
-}
-
-// Acknowledge a ReceivedMessage with the given disposition code.
-func (rm *ReceivedMessage) Acknowledge(disposition Disposition) error {
- return rm.receiver.(*receiver).engine().InjectWait(func() error {
- // Settle doesn't return an error but if the receiver is broken the settlement won't happen.
- rm.eDelivery.SettleAs(uint64(disposition))
- return rm.receiver.Error()
- })
-}
-
-// Accept is short for Acknowledge(Accpeted)
-func (rm *ReceivedMessage) Accept() error { return rm.Acknowledge(Accepted) }
-
-// Reject is short for Acknowledge(Rejected)
-func (rm *ReceivedMessage) Reject() error { return rm.Acknowledge(Rejected) }
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/478ba4ea/proton-c/bindings/go/src/qpid.apache.org/proton/concurrent/sender.go
----------------------------------------------------------------------
diff --git a/proton-c/bindings/go/src/qpid.apache.org/proton/concurrent/sender.go b/proton-c/bindings/go/src/qpid.apache.org/proton/concurrent/sender.go
deleted file mode 100644
index 7a65a24..0000000
--- a/proton-c/bindings/go/src/qpid.apache.org/proton/concurrent/sender.go
+++ /dev/null
@@ -1,190 +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 concurrent
-
-// #include <proton/disposition.h>
-import "C"
-
-import (
- "qpid.apache.org/proton"
- "qpid.apache.org/proton/amqp"
- "qpid.apache.org/proton/internal"
- "time"
-)
-
-type SenderSettings struct {
- LinkSettings
-}
-
-// Sender is a Link that sends messages.
-type Sender interface {
- Link
-
- // Send a message asynchronously, return a SentMessage to identify it.
- //
- // Returns nil if the link is in Unreliable mode and no acknowledgement
- // will be received.
- //
- // See Credit() for note on buffering.
- //
- // Use SentMessage.Disposition() to wait for acknowledgement.
- Send(m amqp.Message) (sm SentMessage, err error)
-}
-
-type sender struct{ link }
-
-func newSender(l link) Sender { return &sender{l} }
-
-// Open the Sender, must be called before calling Send().
-func (s *sender) Open() error {
- err := s.engine().InjectWait(func() error {
- err := s.open()
- if err == nil {
- s.handler().addLink(s.eLink, s)
- }
- return err
- })
- return s.setError(err)
-}
-
-// Disposition indicates the outcome of a settled message delivery.
-type Disposition uint64
-
-const (
- // No disposition available, not yet acknowledged or an error occurred
- NoDisposition Disposition = 0
- // Message was accepted by the receiver
- Accepted = proton.Accepted
- // Message was rejected as invalid by the receiver
- Rejected = proton.Rejected
- // Message was not processed by the receiver but may be processed by some other receiver.
- Released = proton.Released
-)
-
-// String human readable name for a Disposition.
-func (d Disposition) String() string {
- switch d {
- case NoDisposition:
- return "no-disposition"
- case Accepted:
- return "accepted"
- case Rejected:
- return "rejected"
- case Released:
- return "released"
- default:
- return "unknown"
- }
-}
-
-func (s *sender) Send(m amqp.Message) (SentMessage, error) {
- internal.Assert(s.IsOpen(), "sender is not open: %s", s)
- if err := s.Error(); err != nil {
- return nil, err
- }
- var sm SentMessage
- err := s.engine().InjectWait(func() error {
- eDelivery, err := s.eLink.Send(m)
- if err == nil {
- if s.eLink.SndSettleMode() == proton.SndSettled {
- eDelivery.Settle()
- } else {
- sm = newSentMessage(s.session.connection, eDelivery)
- s.session.connection.handler.sentMessages[eDelivery] = sm.(*sentMessage)
- }
- }
- return err
- })
- return sm, err
-}
-
-func (s *sender) closed(err error) {
- s.closeError(err)
-}
-
-// SentMessage represents a previously sent message. It allows you to wait for acknowledgement.
-type SentMessage interface {
- // Disposition blocks till the message is acknowledged and returns the
- // disposition state. NoDisposition means the Connection or the SentMessage
- // was closed before the message was acknowledged.
- Disposition() (Disposition, error)
-
- // DispositionTimeout is like Disposition but gives up after timeout, see Timeout.
- DispositionTimeout(time.Duration) (Disposition, error)
-
- // Forget interrupts any call to Disposition on this SentMessage and tells the
- // peer we are no longer interested in the disposition of this message.
- Forget()
-
- // Error returns the error that closed the disposition, or nil if there was no error.
- // If the disposition closed because the connection closed, it will return Closed.
- Error() error
-}
-
-type sentMessage struct {
- connection *connection
- eDelivery proton.Delivery
- done chan struct{}
- disposition Disposition
- err error
-}
-
-func newSentMessage(c *connection, d proton.Delivery) *sentMessage {
- return &sentMessage{c, d, make(chan struct{}), NoDisposition, nil}
-}
-
-func (sm *sentMessage) Disposition() (Disposition, error) {
- <-sm.done
- return sm.disposition, sm.err
-}
-
-func (sm *sentMessage) DispositionTimeout(timeout time.Duration) (Disposition, error) {
- if _, _, timedout := timedReceive(sm.done, timeout); timedout {
- return sm.disposition, Timeout
- } else {
- return sm.disposition, sm.err
- }
-}
-
-func (sm *sentMessage) Forget() {
- sm.connection.engine.Inject(func() {
- sm.eDelivery.Settle()
- delete(sm.connection.handler.sentMessages, sm.eDelivery)
- })
- sm.finish()
-}
-
-func (sm *sentMessage) settled(err error) {
- if sm.eDelivery.Settled() {
- sm.disposition = Disposition(sm.eDelivery.Remote().Type())
- }
- sm.err = err
- sm.finish()
-}
-
-func (sm *sentMessage) finish() {
- select {
- case <-sm.done: // No-op if already closed
- default:
- close(sm.done)
- }
-}
-
-func (sm *sentMessage) Error() error { return sm.err }
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/478ba4ea/proton-c/bindings/go/src/qpid.apache.org/proton/concurrent/session.go
----------------------------------------------------------------------
diff --git a/proton-c/bindings/go/src/qpid.apache.org/proton/concurrent/session.go b/proton-c/bindings/go/src/qpid.apache.org/proton/concurrent/session.go
deleted file mode 100644
index 2f609be..0000000
--- a/proton-c/bindings/go/src/qpid.apache.org/proton/concurrent/session.go
+++ /dev/null
@@ -1,114 +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 concurrent
-
-import (
- "qpid.apache.org/proton"
- "qpid.apache.org/proton/internal"
-)
-
-// Session is an AMQP session, it contains Senders and Receivers.
-//
-type Session interface {
- Endpoint
-
- // Connection owning this session.
- Connection() Connection
-
- // Sender opens a new sender. v can be a string, which is used as the Target
- // address, or a SenderSettings struct containing more details settings.
- Sender(v interface{}) (Sender, error)
-
- // Receiver opens a new Receiver. v can be a string, which is used as the
- // Source address, or a ReceiverSettings struct containing more details
- // settings.
- Receiver(v interface{}) (Receiver, error)
-}
-
-type session struct {
- endpoint
- eSession proton.Session
- connection *connection
-}
-
-// in proton goroutine
-func newSession(c *connection, es proton.Session) *session {
- return &session{
- connection: c,
- eSession: es,
- endpoint: endpoint{str: es.String()},
- }
-}
-
-func (s *session) Connection() Connection { return s.connection }
-func (s *session) eEndpoint() proton.Endpoint { return s.eSession }
-func (s *session) engine() *proton.Engine { return s.connection.engine }
-func (s *session) Open() error { s.engine().Inject(s.eSession.Open); return nil }
-func (s *session) Close(err error) {
- s.engine().Inject(func() { localClose(s.eSession, err) })
-}
-
-func (s *session) Sender(v interface{}) (snd Sender, err error) {
- var settings LinkSettings
- switch v := v.(type) {
- case string:
- settings.Target = v
- case SenderSettings:
- settings = v.LinkSettings
- default:
- internal.Assert(false, "NewSender() want string or SenderSettings, got %T", v)
- }
- err = s.engine().InjectWait(func() error {
- l, err := makeLocalLink(s, true, settings)
- snd = newSender(l)
- return err
- })
- if err == nil {
- err = snd.Open()
- }
- return
-}
-
-func (s *session) Receiver(v interface{}) (rcv Receiver, err error) {
- var settings ReceiverSettings
- switch v := v.(type) {
- case string:
- settings.Source = v
- case ReceiverSettings:
- settings = v
- default:
- internal.Assert(false, "NewReceiver() want string or ReceiverSettings, got %T", v)
- }
- err = s.engine().InjectWait(func() error {
- l, err := makeLocalLink(s, false, settings.LinkSettings)
- rcv = newReceiver(l)
- return err
- })
- rcv.SetCapacity(settings.Capacity, settings.Prefetch)
- if err == nil {
- err = rcv.Open()
- }
- return
-}
-
-// Called from handler on closed.
-func (s *session) closed(err error) {
- s.closeError(err)
-}
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/478ba4ea/proton-c/bindings/go/src/qpid.apache.org/proton/concurrent/time.go
----------------------------------------------------------------------
diff --git a/proton-c/bindings/go/src/qpid.apache.org/proton/concurrent/time.go b/proton-c/bindings/go/src/qpid.apache.org/proton/concurrent/time.go
deleted file mode 100644
index e9093d3..0000000
--- a/proton-c/bindings/go/src/qpid.apache.org/proton/concurrent/time.go
+++ /dev/null
@@ -1,71 +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 concurrent
-
-import (
- "qpid.apache.org/proton/internal"
- "reflect"
- "time"
-)
-
-// Timeout is the error returned if an operation does not complete on time.
-//
-// Methods named *Timeout in this package take time.Duration timeout parameter.
-//
-// If timeout > 0 and there is no result available before the timeout, they
-// return a zero or nil value and Timeout as an error.
-//
-// If timeout == 0 they will return a result if one is immediatley available or
-// nil/zero and Timeout as an error if not.
-//
-// If timeout == Forever the function will return only when there is a result or
-// some non-timeout error occurs.
-//
-var Timeout = internal.Errorf("timeout")
-
-// Forever can be used as a timeout parameter to indicate wait forever.
-const Forever time.Duration = -1
-
-// timedReceive receives on channel (which can be a chan of any type), waiting
-// up to timeout.
-//
-// timeout==0 means do a non-blocking receive attempt. timeout < 0 means block
-// forever. Other values mean block up to the timeout.
-//
-func timedReceive(channel interface{}, timeout time.Duration) (value interface{}, ok bool, timedout bool) {
- cases := []reflect.SelectCase{
- reflect.SelectCase{Dir: reflect.SelectRecv, Chan: reflect.ValueOf(channel)},
- }
- switch {
- case timeout == 0: // Non-blocking receive
- cases = append(cases, reflect.SelectCase{Dir: reflect.SelectDefault})
- case timeout < 0: // Block forever, nothing to add
- default: // Block up to timeout
- cases = append(cases,
- reflect.SelectCase{Dir: reflect.SelectRecv, Chan: reflect.ValueOf(time.After(timeout))})
- }
- chosen, recv, recvOk := reflect.Select(cases)
- switch {
- case chosen == 0:
- return recv.Interface(), recvOk, false
- default:
- return nil, false, true
- }
-}
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/478ba4ea/proton-c/bindings/go/src/qpid.apache.org/proton/doc.go
----------------------------------------------------------------------
diff --git a/proton-c/bindings/go/src/qpid.apache.org/proton/doc.go b/proton-c/bindings/go/src/qpid.apache.org/proton/doc.go
index 25b43af..e9d6d6f 100644
--- a/proton-c/bindings/go/src/qpid.apache.org/proton/doc.go
+++ b/proton-c/bindings/go/src/qpid.apache.org/proton/doc.go
@@ -18,19 +18,14 @@ under the License.
*/
/*
+Package proton is an event-driven, concurrent-unsafe Go library for AMQP messaging.
+You can write clients and servers using this library.
-Package proton is a Go binding for the Qpid Proton AMQP messaging toolkit (see
-http://qpid.apache.org/proton) It is a concurrent-unsafe, event-driven API that
-closely follows the Proton C API.
-
-Package qpid.apache.org/proton/concurrent provides an alternative,
-concurrent-safe, procedural API. Most applications will find the concurrent API
-easier to use.
-
-If you need direct access to the underlying proton library for some reason, this
-package provides it. The types in this package are simple wrappers for C
-pointers. They provide access to C functions as Go methods and do some trivial
-conversions, for example between Go string and C null-terminated char* strings.
+This package is a port of the Proton C API into Go (see
+http://qpid.apache.org/proton) Go programmers may find the 'electron' package
+more convenient, it provides a concurrent-safe API that allows you to do
+procedural loops in goroutines rather than implementing event handlers that must
+run in a single goroutine.
Consult the C API documentation at http://qpid.apache.org/proton for more
information about the types here. There is a 1-1 correspondence between C type
@@ -42,22 +37,26 @@ and Go method
func (proton.Foo) DoSomething(...)
-The proton.Engine type pumps data between a Go net.Conn connection and a
-proton.Connection goroutine that feeds events to a proton.MessagingHandler. See
-the proton.Engine documentation for more detail.
+The proton.Engine type pumps data between a Go net.Conn and a proton event loop
+goroutine that feeds events to a proton.MessagingHandler, which you must implement.
+See the Engine documentation for more.
+
+MessagingHandler defines an event handling interface that you can implement to
+react to AMQP protocol events. (There is also a lower-level EventHandler, but
+MessagingHandler provides a simpler set of events and automates common tasks for you.)
-EventHandler and MessagingHandler define an event handling interfaces that you
-can implement to react to protocol events. MessagingHandler provides a somewhat
-simpler set of events and automates some common tasks for you.
+All events generated by proton are handled in the single event-loop goroutine
+associated with the Connection and Engine. You can use Engine.Inject() or
+Engine.InjectWait() to inject additional functions into the event loop. Only
+injected functions or handler functions can use proton types (such as Session,
+Link etc.) Handlers and injected functions can set up channels to communicate
+with other goroutines..
-You must ensure that all events are handled in a single goroutine or that you
-serialize all all uses of the proton objects associated with a single connection
-using a lock. You can use channels to communicate between application
-goroutines and the event-handling goroutine, see Engine documentation for more details.
+Separate Connection and Engine instances are independent, and can run concurrently.
-Package qpid.apache.org/proton/concurrent does all this for you and presents a
-simple concurrent-safe interface, for most applications you should use that
-instead.
+The 'electron' package is built on the proton package but instead offers a
+concurrent-safe API that can use simple procedural loops rather than event
+handlers to express application logic. It may be easier to use w for some applications.
*/
package proton
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org