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/09/28 23:23:07 UTC
[23/36] qpid-proton git commit: PROTON-827: Simplified examples and
Connection error handling.
PROTON-827: Simplified examples and Connection error handling.
- Simplified non-relevant code in examples (logging, argument handling)
- Improved error handling on API see Connection.Error(). Need more on other endpoints.
- Added -debug flag to example_test to help debug example problems.
Project: http://git-wip-us.apache.org/repos/asf/qpid-proton/repo
Commit: http://git-wip-us.apache.org/repos/asf/qpid-proton/commit/495d7413
Tree: http://git-wip-us.apache.org/repos/asf/qpid-proton/tree/495d7413
Diff: http://git-wip-us.apache.org/repos/asf/qpid-proton/diff/495d7413
Branch: refs/heads/master
Commit: 495d74136f165f37f692e68710a1b6cdbd74dc98
Parents: 7cecb8e
Author: Alan Conway <ac...@redhat.com>
Authored: Mon May 25 17:14:22 2015 -0400
Committer: Alan Conway <ac...@redhat.com>
Committed: Mon Sep 28 17:19:56 2015 -0400
----------------------------------------------------------------------
examples/go/example_test.go | 30 ++--
examples/go/receive.go | 140 +++++++------------
examples/go/send.go | 108 ++++++--------
proton-c/bindings/go/genwrap.go | 108 +++++++-------
.../src/qpid.apache.org/proton/go/event/pump.go | 5 +-
.../qpid.apache.org/proton/go/internal/error.go | 25 ++--
.../proton/go/messaging/messaging.go | 51 ++++---
7 files changed, 215 insertions(+), 252 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/495d7413/examples/go/example_test.go
----------------------------------------------------------------------
diff --git a/examples/go/example_test.go b/examples/go/example_test.go
index a4b4c2c..8879c38 100644
--- a/examples/go/example_test.go
+++ b/examples/go/example_test.go
@@ -78,13 +78,11 @@ func (b *broker) check() error {
// Start the demo broker, wait till it is listening on *addr. No-op if already started.
func (b *broker) start() error {
- build("event_broker.go")
if b.cmd == nil { // Not already started
// FIXME aconway 2015-04-30: better way to pick/configure a broker port.
b.addr = fmt.Sprintf("127.0.0.1:%d", rand.Intn(10000)+10000)
- b.cmd = exec.Command(exepath("event_broker"), "-addr", b.addr, "-verbose", "0")
+ b.cmd = exampleCommand("event_broker", "-addr", b.addr)
b.runerr = make(chan error)
- // Change the -verbose setting above to see broker output on stdout/stderr.
b.cmd.Stderr, b.cmd.Stdout = os.Stderr, os.Stdout
go func() {
b.runerr <- b.cmd.Run()
@@ -111,7 +109,12 @@ func checkEqual(want interface{}, got interface{}) error {
// runCommand returns an exec.Cmd to run an example.
func exampleCommand(prog string, arg ...string) *exec.Cmd {
build(prog + ".go")
- cmd := exec.Command(exepath(prog), arg...)
+ args := []string{}
+ if *debug {
+ args = append(args, "-debug=true")
+ }
+ args = append(args, arg...)
+ cmd := exec.Command(exepath(prog), args...)
cmd.Stderr = os.Stderr
return cmd
}
@@ -149,16 +152,16 @@ func TestExampleSendReceive(t *testing.T) {
}
testBroker.start()
err := runExampleWant(
- "send: Received all 15 acknowledgements\n",
+ "Received all 15 acknowledgements\n",
"send",
- exampleArgs("-count", "5", "-verbose", "1")...)
+ exampleArgs("-count", "5")...)
if err != nil {
t.Fatal(err)
}
err = runExampleWant(
- "receive: Listening\nreceive: Received 15 messages\n",
+ "Listening on 3 connections\nReceived 15 messages\n",
"receive",
- exampleArgs("-verbose", "1", "-count", "15")...)
+ exampleArgs("-count", "15")...)
if err != nil {
t.Fatal(err)
}
@@ -187,7 +190,7 @@ func goReceiveWant(errchan chan<- error, want string, arg ...string) *exec.Cmd {
errchan <- err
return
}
- listening := "receive: Listening\n"
+ listening := "Listening on 3 connections\n"
if line != listening {
errchan <- checkEqual(listening, line)
return
@@ -209,8 +212,8 @@ func TestExampleReceiveSend(t *testing.T) {
testBroker.start()
recvErr := make(chan error)
recvCmd := goReceiveWant(recvErr,
- "receive: Received 15 messages\n",
- exampleArgs("-count", "15", "-verbose", "1")...)
+ "Received 15 messages\n",
+ exampleArgs("-count", "15")...)
defer func() {
recvCmd.Process.Kill()
recvCmd.Wait()
@@ -219,9 +222,9 @@ func TestExampleReceiveSend(t *testing.T) {
t.Fatal(err)
}
err := runExampleWant(
- "send: Received all 15 acknowledgements\n",
+ "Received all 15 acknowledgements\n",
"send",
- exampleArgs("-count", "5", "-verbose", "1")...)
+ exampleArgs("-count", "5")...)
if err != nil {
t.Fatal(err)
}
@@ -263,6 +266,7 @@ func build(prog string) {
}
var rpath = flag.String("rpath", "", "Runtime path for test executables")
+var debug = flag.Bool("debug", false, "Debugging output from examples")
func TestMain(m *testing.M) {
rand.Seed(time.Now().UTC().UnixNano())
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/495d7413/examples/go/receive.go
----------------------------------------------------------------------
diff --git a/examples/go/receive.go b/examples/go/receive.go
index e31862b..b1eb309 100644
--- a/examples/go/receive.go
+++ b/examples/go/receive.go
@@ -22,87 +22,65 @@ package main
import (
"flag"
"fmt"
- "io"
- "io/ioutil"
- "log"
- "math"
"net"
"os"
- "path"
"qpid.apache.org/proton/go/amqp"
"qpid.apache.org/proton/go/messaging"
"sync"
- "time"
)
-// Command-line flags
-var verbose = flag.Int("verbose", 1, "Output level, 0 means none, higher means more")
-var count = flag.Int64("count", 0, "Stop after receiving this many messages. 0 means unlimited.")
-var timeout = flag.Int64("time", 0, "Stop after this many seconds. 0 means unlimited.")
+// Usage and command-line flags
+func usage() {
+ fmt.Fprintf(os.Stderr, `Usage: %s url [url ...]
+Receive messages from all the listed URLs concurrently and print them.
+`, os.Args[0])
+ flag.PrintDefaults()
+}
+
+var debug = flag.Bool("debug", false, "Print detailed debug output")
+var count = flag.Uint64("count", 1, "Stop after receiving this many messages.")
var full = flag.Bool("full", false, "Print full message not just body.")
func main() {
- // Parse flags and arguments, print usage message on error.
- flag.Usage = func() {
- fmt.Fprintf(os.Stderr, `
-Usage: %s url [url ...]
-Receive messages from all the listed URLs concurrently and print them.
-`, os.Args[0])
- flag.PrintDefaults()
- }
+ flag.Usage = usage
flag.Parse()
+
urls := flag.Args() // Non-flag arguments are URLs to receive from
if len(urls) == 0 {
- flag.Usage()
- fmt.Fprintf(os.Stderr, "No URL provided")
+ fmt.Fprintln(os.Stderr, "No URL provided")
+ usage()
os.Exit(1)
}
- duration := time.Duration(*timeout) * time.Second
- if duration == 0 {
- duration = time.Duration(math.MaxInt64) // Not forever, but 290 years is close enough.
- }
- if *count == 0 {
- *count = math.MaxInt64
- }
-
- // Create a goroutine for each URL that receives messages and sends them to
- // the messages channel. main() receives and prints them.
messages := make(chan amqp.Message) // Channel for messages from goroutines to main()
stop := make(chan struct{}) // Closing this channel means the program is stopping.
+ var wait sync.WaitGroup // Used by main() to wait for all goroutines to end.
+ wait.Add(len(urls)) // Wait for one goroutine per URL.
- var wait sync.WaitGroup // Used by main() to wait for all goroutines to end.
-
- wait.Add(len(urls)) // Wait for one goroutine per URL.
-
- // Arrange to close all connections on exit
- connections := make([]*messaging.Connection, len(urls))
- defer func() {
- for _, c := range connections {
- if c != nil {
- c.Close()
- }
- }
- }()
+ connections := make([]*messaging.Connection, len(urls)) // Store connctions to close on exit
+ // Start a goroutine to for each URL to receive messages and send them to the messages channel.
+ // main() receives and prints them.
for i, urlStr := range urls {
- debug.Printf("Connecting to %s", urlStr)
- go func(urlStr string) {
- defer wait.Done() // Notify main() that this goroutine is done.
+ debugf("debug: Connecting to %s\n", urlStr)
+ go func(urlStr string) { // Start the goroutine
+
+ defer wait.Done() // Notify main() when this goroutine is done.
url, err := amqp.ParseURL(urlStr) // Like net/url.Parse() but with AMQP defaults.
- fatalIf(err)
+ exitIf(err)
// Open a standard Go net.Conn and and AMQP connection using it.
conn, err := net.Dial("tcp", url.Host) // Note net.URL.Host is actually "host:port"
- fatalIf(err)
+ exitIf(err)
pc, err := messaging.Connect(conn) // This is our AMQP connection.
- fatalIf(err)
- connections[i] = pc // So we can close it when main() ends
+ exitIf(err)
+ connections[i] = pc // Save connection so it will be closed when main() ends
// Create a receiver using the path of the URL as the AMQP address
r, err := pc.Receiver(url.Path)
- fatalIf(err)
+ exitIf(err)
+ // Loop receiving messages
for {
var m amqp.Message
select { // Receive a message or stop.
@@ -118,57 +96,43 @@ Receive messages from all the listed URLs concurrently and print them.
}
}(urlStr)
}
- info.Printf("Listening")
-
- // time.After() returns a channel that will close when the timeout is up.
- timer := time.After(duration)
-
- // main() prints each message and checks for count or timeout being exceeded.
- for i := int64(0); i < *count; i++ {
- select {
- case m := <-messages:
- debug.Print(formatMessage{m})
- case <-timer: // Timeout has expired
- i = 0
- }
+
+ // All goroutines are started, we are receiving messages.
+ fmt.Printf("Listening on %d connections\n", len(urls))
+
+ // print each message until the count is exceeded.
+ for i := uint64(0); i < *count; i++ {
+ debugf("%s\n", formatMessage(<-messages))
}
- info.Printf("Received %d messages", *count)
+ fmt.Printf("Received %d messages\n", *count)
close(stop) // Signal all goroutines to stop.
wait.Wait() // Wait for all goroutines to finish.
-}
-
-// Logging
-func logger(prefix string, level int, w io.Writer) *log.Logger {
- if *verbose >= level {
- return log.New(w, prefix, 0)
+ close(messages)
+ for _, c := range connections { // Close all connections
+ if c != nil {
+ c.Close()
+ }
}
- return log.New(ioutil.Discard, "", 0)
}
-var info, debug *log.Logger
-
-func init() {
- flag.Parse()
- name := path.Base(os.Args[0])
- log.SetFlags(0) // Use default logger for errors.
- log.SetPrefix(fmt.Sprintf("%s: ", name)) // Log errors on stderr.
- info = logger(fmt.Sprintf("%s: ", name), 1, os.Stdout) // Log info on stdout.
- debug = logger(fmt.Sprintf("%s debug: ", name), 2, os.Stderr) // Log debug on stderr.
+// Simple debug logging
+func debugf(format string, data ...interface{}) {
+ if *debug {
+ fmt.Fprintf(os.Stderr, format, data...)
+ }
}
// Simple error handling for demo.
-func fatalIf(err error) {
+func exitIf(err error) {
if err != nil {
- log.Fatal(err)
+ fmt.Fprintln(os.Stderr, err)
}
}
-type formatMessage struct{ m amqp.Message }
-
-func (fm formatMessage) String() string {
+func formatMessage(m amqp.Message) string {
if *full {
- return fmt.Sprintf("%#v", fm.m)
+ return fmt.Sprintf("%#v", m)
} else {
- return fmt.Sprintf("%#v", fm.m.Body())
+ return fmt.Sprintf("%#v", m.Body())
}
}
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/495d7413/examples/go/send.go
----------------------------------------------------------------------
diff --git a/examples/go/send.go b/examples/go/send.go
index 4aaeb43..98acefa 100644
--- a/examples/go/send.go
+++ b/examples/go/send.go
@@ -22,21 +22,23 @@ package main
import (
"flag"
"fmt"
- "io"
- "io/ioutil"
- "log"
- "math"
"net"
"os"
- "path"
"qpid.apache.org/proton/go/amqp"
"qpid.apache.org/proton/go/messaging"
"sync"
)
-// Command-line flags
-var verbose = flag.Int("verbose", 1, "Output level, 0 means none, higher means more")
-var count = flag.Int64("count", 1, "Send this may messages per address. 0 means unlimited.")
+// Usage and command-line flags
+func usage() {
+ fmt.Fprintf(os.Stderr, `Usage: %s url [url ...]
+Send messages to each URL concurrently with body "<url-path>-<n>" where n is the message number.
+`, os.Args[0])
+ flag.PrintDefaults()
+}
+
+var debug = flag.Bool("debug", false, "Print detailed debug output")
+var count = flag.Int64("count", 1, "Send this may messages per address.")
// Ack associates an info string with an acknowledgement
type Ack struct {
@@ -45,67 +47,51 @@ type Ack struct {
}
func main() {
- // Parse flags and arguments, print usage message on error.
- flag.Usage = func() {
- fmt.Fprintf(os.Stderr, `
-Usage: %s url [url ...]
-Send messages to all the listed URLs concurrently.
-To each URL, send the string "path-n" where n is the message number.
-`, os.Args[0])
- flag.PrintDefaults()
- }
+ flag.Usage = usage
flag.Parse()
+
urls := flag.Args() // Non-flag arguments are URLs to receive from
if len(urls) == 0 {
+ fmt.Fprintln(os.Stderr, "No URL provided")
flag.Usage()
- fmt.Fprintf(os.Stderr, "No URL provided\n")
os.Exit(1)
}
- if *count == 0 {
- *count = math.MaxInt64
- }
-
- // Create a channel to receive all the acknowledgements
- acks := make(chan Ack)
- // Create a goroutine for each URL that sends messages.
+ acks := make(chan Ack) // Channel to receive all the acknowledgements
var wait sync.WaitGroup // Used by main() to wait for all goroutines to end.
wait.Add(len(urls)) // Wait for one goroutine per URL.
- // Arrange to close all connections on exit
- connections := make([]*messaging.Connection, len(urls))
- defer func() {
- for _, c := range connections {
- if c != nil {
- c.Close()
- }
- }
- }()
+ connections := make([]*messaging.Connection, len(urls)) // Store connctions to close on exit
+ // Start a goroutine for each URL to send messages, receive the acknowledgements and
+ // send them to the acks channel.
for i, urlStr := range urls {
- debug.Printf("Connecting to %v", urlStr)
+ debugf("Connecting to %v\n", urlStr)
go func(urlStr string) {
+
defer wait.Done() // Notify main() that this goroutine is done.
url, err := amqp.ParseURL(urlStr) // Like net/url.Parse() but with AMQP defaults.
- fatalIf(err)
+ exitIf(err)
// Open a standard Go net.Conn and and AMQP connection using it.
conn, err := net.Dial("tcp", url.Host) // Note net.URL.Host is actually "host:port"
- fatalIf(err)
+ exitIf(err)
pc, err := messaging.Connect(conn) // This is our AMQP connection.
- fatalIf(err)
- connections[i] = pc // So we can close it when main() ends
+ exitIf(err)
+ connections[i] = pc // Save connection so it will be closed when main() ends
// Create a sender using the path of the URL as the AMQP address
s, err := pc.Sender(url.Path)
- fatalIf(err)
+ exitIf(err)
+ // Loop sending messages, receiving acknowledgements and sending them to the acks channel.
for i := int64(0); i < *count; i++ {
m := amqp.NewMessage()
body := fmt.Sprintf("%v-%v", url.Path, i)
m.SetBody(body)
+ // Note Send is *asynchronous*, ack is a channel that will receive the acknowledgement.
ack, err := s.Send(m)
- fatalIf(err)
+ exitIf(err)
acks <- Ack{ack, body} // Send the acknowledgement to main()
}
}(urlStr)
@@ -113,44 +99,38 @@ To each URL, send the string "path-n" where n is the message number.
// Wait for all the acknowledgements
expect := int(*count) * len(urls)
- debug.Printf("Started senders, expect %v acknowledgements", expect)
+ debugf("Started senders, expect %v acknowledgements\n", expect)
for i := 0; i < expect; i++ {
ack, ok := <-acks
if !ok {
- info.Fatalf("acks channel closed after only %d acks\n", i)
+ exitIf(fmt.Errorf("acks channel closed after only %d acks\n", i))
}
d := <-ack.ack
- debug.Printf("acknowledgement[%v] %v", i, ack.info)
+ debugf("acknowledgement[%v] %v\n", i, ack.info)
if d != messaging.Accepted {
- info.Printf("Unexpected disposition %v", d)
+ fmt.Printf("Unexpected disposition %v\n", d)
}
}
- info.Printf("Received all %v acknowledgements", expect)
- wait.Wait() // Wait for all goroutines to finish.
-}
+ fmt.Printf("Received all %v acknowledgements\n", expect)
-// Logging
-func logger(prefix string, level int, w io.Writer) *log.Logger {
- if *verbose >= level {
- return log.New(w, prefix, 0)
+ wait.Wait() // Wait for all goroutines to finish.
+ for _, c := range connections { // Close all connections
+ if c != nil {
+ c.Close()
+ }
}
- return log.New(ioutil.Discard, "", 0)
}
-var info, debug *log.Logger
-
-func init() {
- flag.Parse()
- name := path.Base(os.Args[0])
- log.SetFlags(0) // Use default logger for errors.
- log.SetPrefix(fmt.Sprintf("%s: ", name)) // Log errors on stderr.
- info = logger(fmt.Sprintf("%s: ", name), 1, os.Stdout) // Log info on stdout.
- debug = logger(fmt.Sprintf("%s debug: ", name), 2, os.Stderr) // Log debug on stderr.
+// Simple debug logging
+func debugf(format string, data ...interface{}) {
+ if *debug {
+ fmt.Fprintf(os.Stderr, format, data...)
+ }
}
// Simple error handling for demo.
-func fatalIf(err error) {
+func exitIf(err error) {
if err != nil {
- log.Fatal(err)
+ fmt.Fprintln(os.Stderr, err)
}
}
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/495d7413/proton-c/bindings/go/genwrap.go
----------------------------------------------------------------------
diff --git a/proton-c/bindings/go/genwrap.go b/proton-c/bindings/go/genwrap.go
index 094b196..27e5966 100644
--- a/proton-c/bindings/go/genwrap.go
+++ b/proton-c/bindings/go/genwrap.go
@@ -35,6 +35,59 @@ import (
"text/template"
)
+var includeProton = "../../include/proton"
+var outpath = "src/qpid.apache.org/proton/go/event/wrappers_gen.go"
+
+func main() {
+ flag.Parse()
+ out, err := os.Create(outpath)
+ panicIf(err)
+ defer out.Close()
+
+ apis := []string{"session", "link", "delivery", "disposition", "condition", "terminus", "connection"}
+ fmt.Fprintln(out, copyright)
+ fmt.Fprint(out, `
+package event
+
+import (
+ "time"
+ "unsafe"
+ "qpid.apache.org/proton/go/internal"
+)
+
+// #include <proton/types.h>
+// #include <proton/event.h>
+// #include <stdlib.h>
+`)
+ for _, api := range apis {
+ fmt.Fprintf(out, "// #include <proton/%s.h>\n", api)
+ }
+ fmt.Fprintln(out, `import "C"`)
+
+ event(out)
+
+ for _, api := range apis {
+ fmt.Fprintf(out, "// Wrappers for declarations in %s.h\n\n", api)
+ header := readHeader(api)
+ enums := findEnums(header)
+ for _, e := range enums {
+ genEnum(out, e.Name, e.Values)
+ }
+ apiWrapFns(api, header, out)
+ }
+ out.Close()
+
+ // Run gofmt.
+ cmd := exec.Command("gofmt", "-w", outpath)
+ cmd.Stdout = os.Stdout
+ cmd.Stderr = os.Stderr
+ err = cmd.Run()
+ if err != nil {
+ fmt.Fprintf(os.Stderr, "gofmt: %s", err)
+ os.Exit(1)
+ }
+}
+
func mixedCase(s string) string {
result := ""
for _, w := range strings.Split(s, "_") {
@@ -96,7 +149,7 @@ func panicIf(err error) {
}
func readHeader(name string) string {
- file, err := os.Open(path.Join(*includeProton, name+".h"))
+ file, err := os.Open(path.Join(includeProton, name+".h"))
panicIf(err)
defer file.Close()
s, err := ioutil.ReadAll(file)
@@ -372,56 +425,3 @@ func apiWrapFns(api, header string, out io.Writer) {
fmt.Fprintf(out, "}\n")
}
}
-
-var includeProton = flag.String("include", "", "path to proton include files, including /proton")
-
-func main() {
- flag.Parse()
- outpath := "wrappers_gen.go"
- out, err := os.Create(outpath)
- panicIf(err)
- defer out.Close()
-
- apis := []string{"session", "link", "delivery", "disposition", "condition", "terminus", "connection"}
- fmt.Fprintln(out, copyright)
- fmt.Fprint(out, `
-package event
-
-import (
- "time"
- "unsafe"
- "qpid.apache.org/proton/go/internal"
-)
-
-// #include <proton/types.h>
-// #include <proton/event.h>
-// #include <stdlib.h>
-`)
- for _, api := range apis {
- fmt.Fprintf(out, "// #include <proton/%s.h>\n", api)
- }
- fmt.Fprintln(out, `import "C"`)
-
- event(out)
-
- for _, api := range apis {
- fmt.Fprintf(out, "// Wrappers for declarations in %s.h\n\n", api)
- header := readHeader(api)
- enums := findEnums(header)
- for _, e := range enums {
- genEnum(out, e.Name, e.Values)
- }
- apiWrapFns(api, header, out)
- }
- out.Close()
-
- // Run gofmt.
- cmd := exec.Command("gofmt", "-w", outpath)
- cmd.Stdout = os.Stdout
- cmd.Stderr = os.Stderr
- err = cmd.Run()
- if err != nil {
- fmt.Fprintf(os.Stderr, "gofmt: %s", err)
- os.Exit(1)
- }
-}
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/495d7413/proton-c/bindings/go/src/qpid.apache.org/proton/go/event/pump.go
----------------------------------------------------------------------
diff --git a/proton-c/bindings/go/src/qpid.apache.org/proton/go/event/pump.go b/proton-c/bindings/go/src/qpid.apache.org/proton/go/event/pump.go
index db022ff..73db513 100644
--- a/proton-c/bindings/go/src/qpid.apache.org/proton/go/event/pump.go
+++ b/proton-c/bindings/go/src/qpid.apache.org/proton/go/event/pump.go
@@ -101,7 +101,7 @@ you are doing something fairly low-level it is probably a better choice.
*/
type Pump struct {
// Error is set on exit from Run() if there was an error.
- Error error
+ Error error // FIXME aconway 2015-05-26: make it a function
// Channel to inject functions to be executed in the Pump's proton event loop.
Inject chan func()
@@ -212,6 +212,9 @@ func (p *Pump) Close() error {
}
delete(pumps, p.connection)
p.free()
+ if p.Error == io.EOF {
+ return nil
+ }
return p.Error
}
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/495d7413/proton-c/bindings/go/src/qpid.apache.org/proton/go/internal/error.go
----------------------------------------------------------------------
diff --git a/proton-c/bindings/go/src/qpid.apache.org/proton/go/internal/error.go b/proton-c/bindings/go/src/qpid.apache.org/proton/go/internal/error.go
index 01ba890..f3f3307 100644
--- a/proton-c/bindings/go/src/qpid.apache.org/proton/go/internal/error.go
+++ b/proton-c/bindings/go/src/qpid.apache.org/proton/go/internal/error.go
@@ -29,7 +29,6 @@ import (
"fmt"
"runtime"
"sync"
- "sync/atomic"
"unsafe"
)
@@ -105,21 +104,23 @@ func panicIf(condition bool, fmt string, args ...interface{}) {
// FirstError is a goroutine-safe error holder that keeps the first error that is set.
type FirstError struct {
- err atomic.Value
- once sync.Once
+ err error
+ lock sync.Mutex
}
-// Set the error if not allread set.
-func (e *FirstError) Set(err error) {
- e.once.Do(func() { e.err.Store(err) })
+// Set the error if not already set, return the error.
+func (e *FirstError) Set(err error) error {
+ e.lock.Lock()
+ defer e.lock.Unlock()
+ if e.err == nil {
+ e.err = err
+ }
+ return e.err
}
// Get the error.
func (e *FirstError) Get() error {
- v := e.err.Load()
- if v != nil {
- return v.(error)
- } else {
- return nil
- }
+ e.lock.Lock()
+ defer e.lock.Unlock()
+ return e.err
}
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/495d7413/proton-c/bindings/go/src/qpid.apache.org/proton/go/messaging/messaging.go
----------------------------------------------------------------------
diff --git a/proton-c/bindings/go/src/qpid.apache.org/proton/go/messaging/messaging.go b/proton-c/bindings/go/src/qpid.apache.org/proton/go/messaging/messaging.go
index e653de2..e4b117d 100644
--- a/proton-c/bindings/go/src/qpid.apache.org/proton/go/messaging/messaging.go
+++ b/proton-c/bindings/go/src/qpid.apache.org/proton/go/messaging/messaging.go
@@ -23,11 +23,16 @@ package messaging
import "C"
import (
+ "io"
"net"
"qpid.apache.org/proton/go/amqp"
"qpid.apache.org/proton/go/event"
+ "qpid.apache.org/proton/go/internal"
)
+// Closed is an alias for io.EOF. It indicates orderly closure of an endpoint.
+var Closed = io.EOF
+
// Connection is a connection to a remote AMQP endpoint.
//
// You can set exported fields to configure the connection before calling
@@ -42,27 +47,27 @@ type Connection struct {
handler *handler
pump *event.Pump
session Session
+ err internal.FirstError
}
+// Error returns nil if the connection is open, messaging.Closed if was closed cleanly
+// or an error value if it was closed due to an error.
+func (c *Connection) Error() error { return c.err.Get() }
+
// Make an AMQP connection over a net.Conn connection.
-//
-// Use Connection.Close() to close the Connection, this will also close conn.
-// Using conn.Close() directly will cause an abrupt disconnect rather than an
-// orderly AMQP close.
-//
+// You must call c.Close() to close the connection and clean up its resources.
func (c *Connection) Open(conn net.Conn) (err error) {
c.handler = newHandler(c)
c.pump, err = event.NewPump(conn,
event.NewMessagingDelegator(c.handler),
)
- if err != nil {
- return err
- }
- if c.Server {
- c.pump.Server()
+ if err == nil {
+ if c.Server {
+ c.pump.Server()
+ }
+ go c.pump.Run()
}
- go c.pump.Run()
- return nil
+ return c.err.Set(err)
}
// Connect opens a default client connection. It is a shortcut for
@@ -71,14 +76,16 @@ func (c *Connection) Open(conn net.Conn) (err error) {
//
func Connect(conn net.Conn) (*Connection, error) {
c := &Connection{}
- err := c.Open(conn)
- return c, err
+ c.err.Set(c.Open(conn))
+ return c, c.Error()
}
-// Close the connection.
-//
-// Connections must be closed to clean up resources and stop associated goroutines.
-func (c *Connection) Close() error { return c.pump.Close() }
+// Close cleans up resources and closes the associated net.Conn connection.
+func (c *Connection) Close() error {
+ err := c.pump.Close() // Will be nil on close OK
+ c.err.Set(c.pump.Error) // Will be io.EOF on close OK
+ return err
+}
// DefaultSession returns a default session for the connection.
//
@@ -86,6 +93,9 @@ func (c *Connection) Close() error { return c.pump.Close() }
// Use Session() for more control over creating sessions.
//
func (c *Connection) DefaultSession() (s Session, err error) {
+ if c.Error() != nil {
+ return Session{}, c.Error()
+ }
if c.session.e.IsNil() {
c.session, err = c.Session()
}
@@ -237,14 +247,15 @@ func (s *Sender) Send(m amqp.Message) (ack Acknowledgement, err error) {
// Close the sender.
func (s *Sender) Close() error { return nil } // FIXME aconway 2015-04-27: close/free
-// Receiver receives messages via the channel Receive.
type Receiver struct {
Link
- // Channel of messag
+ // Channel to receive messages. When it closes, check Receiver.Error() for an error.
Receive <-chan amqp.Message
}
// FIXME aconway 2015-04-29: settlement - ReceivedMessage with Settle() method?
+// FIXME aconway 2015-05-25: Close must unblock Receive() calls.
+
// Close the Receiver.
func (r *Receiver) Close() error { return nil } // FIXME aconway 2015-04-29: close/free
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org