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 2016/07/07 17:35:04 UTC
[3/3] qpid-proton git commit: PROTON-1237: go: go binding uses C
pn_connection_engine_t
PROTON-1237: go: go binding uses C pn_connection_engine_t
- modified Go binding implementation to use pn_connection_engine_t instead of
directly driving the engine objects.
Project: http://git-wip-us.apache.org/repos/asf/qpid-proton/repo
Commit: http://git-wip-us.apache.org/repos/asf/qpid-proton/commit/c69cd100
Tree: http://git-wip-us.apache.org/repos/asf/qpid-proton/tree/c69cd100
Diff: http://git-wip-us.apache.org/repos/asf/qpid-proton/diff/c69cd100
Branch: refs/heads/master
Commit: c69cd10099fd5a62677339c45605f80d211e67a4
Parents: 301ce7e
Author: Alan Conway <ac...@redhat.com>
Authored: Thu Jul 7 11:09:31 2016 -0400
Committer: Alan Conway <ac...@redhat.com>
Committed: Thu Jul 7 12:18:29 2016 -0400
----------------------------------------------------------------------
.../qpid.apache.org/electron/messaging_test.go | 5 +-
.../go/src/qpid.apache.org/electron/receiver.go | 6 +-
.../go/src/qpid.apache.org/proton/engine.go | 301 ++++++++-----------
.../go/src/qpid.apache.org/proton/handlers.go | 2 +
4 files changed, 137 insertions(+), 177 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/c69cd100/proton-c/bindings/go/src/qpid.apache.org/electron/messaging_test.go
----------------------------------------------------------------------
diff --git a/proton-c/bindings/go/src/qpid.apache.org/electron/messaging_test.go b/proton-c/bindings/go/src/qpid.apache.org/electron/messaging_test.go
index 0de7d16..4a9e652 100644
--- a/proton-c/bindings/go/src/qpid.apache.org/electron/messaging_test.go
+++ b/proton-c/bindings/go/src/qpid.apache.org/electron/messaging_test.go
@@ -421,7 +421,7 @@ func TestConnectionCloseInterrupt1(t *testing.T) {
pairs.server.Close(want)
for i := 0; i < 3; i++ {
if r := <-results; want != r.err {
- t.Logf("want %v got %v", want, r)
+ t.Errorf("want %v got %v", want, r)
}
}
}
@@ -447,8 +447,7 @@ func TestConnectionCloseInterrupt2(t *testing.T) {
pairs.client.Connection().Close(want)
for i := 0; i < 3; i++ {
if r := <-results; want != r.err {
- // TODO aconway 2015-10-06: Not propagating the correct error, seeing nil.
- t.Logf("want %v got %v", want, r.err)
+ t.Errorf("want %v got %v", want, r.err)
}
}
}
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/c69cd100/proton-c/bindings/go/src/qpid.apache.org/electron/receiver.go
----------------------------------------------------------------------
diff --git a/proton-c/bindings/go/src/qpid.apache.org/electron/receiver.go b/proton-c/bindings/go/src/qpid.apache.org/electron/receiver.go
index 3259eaa..026a385 100644
--- a/proton-c/bindings/go/src/qpid.apache.org/electron/receiver.go
+++ b/proton-c/bindings/go/src/qpid.apache.org/electron/receiver.go
@@ -121,14 +121,13 @@ func (r *receiver) flowTopUp() {
}
}
-// Not claled
func (r *receiver) Receive() (rm ReceivedMessage, err error) {
return r.ReceiveTimeout(Forever)
}
func (r *receiver) ReceiveTimeout(timeout time.Duration) (ReceivedMessage, error) {
assert(r.buffer != nil, "Receiver is not open: %s", r)
- select { // Check for immediate availability
+ select { // Check for immediate availability, avoid caller() inject.
case rm := <-r.buffer:
r.flowTopUp()
return rm, nil
@@ -174,10 +173,11 @@ func (r *receiver) message(delivery proton.Delivery) {
}
func (r *receiver) closed(err error) error {
+ e := r.link.closed(err)
if r.buffer != nil {
close(r.buffer)
}
- return r.link.closed(err)
+ return e
}
// ReceivedMessage contains an amqp.Message and allows the message to be acknowledged.
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/c69cd100/proton-c/bindings/go/src/qpid.apache.org/proton/engine.go
----------------------------------------------------------------------
diff --git a/proton-c/bindings/go/src/qpid.apache.org/proton/engine.go b/proton-c/bindings/go/src/qpid.apache.org/proton/engine.go
index eecda7a..5b9dbf0 100644
--- a/proton-c/bindings/go/src/qpid.apache.org/proton/engine.go
+++ b/proton-c/bindings/go/src/qpid.apache.org/proton/engine.go
@@ -19,18 +19,6 @@ under the License.
package proton
-// #include <proton/connection.h>
-// #include <proton/event.h>
-// #include <proton/error.h>
-// #include <proton/handlers.h>
-// #include <proton/session.h>
-// #include <proton/transport.h>
-// #include <memory.h>
-// #include <stdlib.h>
-//
-// PN_HANDLE(REMOTE_ADDR)
-import "C"
-
import (
"fmt"
"net"
@@ -39,6 +27,19 @@ import (
"unsafe"
)
+/*
+#include <proton/connection.h>
+#include <proton/connection_engine.h>
+#include <proton/event.h>
+#include <proton/error.h>
+#include <proton/handlers.h>
+#include <proton/session.h>
+#include <proton/transport.h>
+#include <memory.h>
+#include <stdlib.h>
+*/
+import "C"
+
// Injecter allows functions to be "injected" into the event-processing loop, to
// be called in the same goroutine as event handlers.
type Injecter interface {
@@ -67,21 +68,6 @@ type Injecter interface {
InjectWait(f func() error) error
}
-// bufferChan manages a pair of ping-pong buffers to pass bytes through a channel.
-type bufferChan struct {
- buffers chan []byte
- buf1, buf2 []byte
-}
-
-func newBufferChan(size int) *bufferChan {
- return &bufferChan{make(chan []byte), make([]byte, size), make([]byte, size)}
-}
-
-func (b *bufferChan) buffer() []byte {
- b.buf1, b.buf2 = b.buf2, b.buf1 // Alternate buffers.
- return b.buf1[:cap(b.buf1)]
-}
-
// Engine reads from a net.Conn, decodes AMQP events and calls the appropriate
// Handler functions sequentially in a single goroutine. Actions taken by
// Handler functions (such as sending messages) are encoded and written to the
@@ -116,15 +102,11 @@ type Engine struct {
err ErrorHolder
inject chan func()
- conn net.Conn
- connection Connection
- transport Transport
- collector *C.pn_collector_t
- read *bufferChan // Read buffers channel.
- write *bufferChan // Write buffers channel.
- handlers []EventHandler // Handlers for proton events.
- running chan struct{} // This channel will be closed when the goroutines are done.
- closeOnce sync.Once
+ conn net.Conn
+ engine C.pn_connection_engine_t
+ handlers []EventHandler // Handlers for proton events.
+ running chan struct{} // This channel will be closed when the goroutines are done.
+ closeOnce sync.Once
}
const bufferSize = 4096
@@ -136,37 +118,45 @@ const bufferSize = 4096
// You can check for errors on Engine.Error.
//
func NewEngine(conn net.Conn, handlers ...EventHandler) (*Engine, error) {
- // Save the connection ID for Connection.String()
eng := &Engine{
- inject: make(chan func()),
- conn: conn,
- transport: Transport{C.pn_transport()},
- connection: Connection{C.pn_connection()},
- collector: C.pn_collector(),
- handlers: handlers,
- read: newBufferChan(bufferSize),
- write: newBufferChan(bufferSize),
- running: make(chan struct{}),
- }
- if eng.transport.IsNil() || eng.connection.IsNil() || eng.collector == nil {
- return nil, fmt.Errorf("failed to allocate engine")
+ inject: make(chan func()),
+ conn: conn,
+ handlers: handlers,
+ running: make(chan struct{}),
}
-
- // TODO aconway 2015-06-25: connection settings for user, password, container etc.
- // before transport.Bind() Set up connection before Engine, allow Engine or Reactor
- // to run connection.
-
- // Unique container-id by default.
- eng.connection.SetContainer(UUID4().String())
- pnErr := eng.transport.Bind(eng.connection)
- if pnErr != 0 {
+ if pnErr := C.pn_connection_engine_init(&eng.engine); pnErr != 0 {
return nil, fmt.Errorf("cannot setup engine: %s", PnErrorCode(pnErr))
}
- C.pn_connection_collect(eng.connection.pn, eng.collector)
- eng.connection.Open()
+ // Unique container-id by default.
+ eng.Connection().SetContainer(UUID4().String()) // FIXME aconway 2016-06-21:
+ eng.Connection().Open()
return eng, nil
}
+// Create a byte slice backed by the memory of a pn_buf_t, no copy.
+// Empty buffer {0,0} returns a nil byte slice.
+func byteSlice(data unsafe.Pointer, size C.size_t) []byte {
+ if data == nil || size == 0 {
+ return nil
+ } else {
+ return (*[1 << 30]byte)(data)[:size:size]
+ }
+}
+
+func (eng *Engine) buffers() ([]byte, []byte) {
+ r := C.pn_connection_engine_read_buffer(&eng.engine)
+ w := C.pn_connection_engine_write_buffer(&eng.engine)
+ return byteSlice(unsafe.Pointer(r.data), r.size), byteSlice(unsafe.Pointer(w.data), w.size)
+}
+
+func (eng *Engine) Connection() Connection {
+ return Connection{C.pn_connection_engine_connection(&eng.engine)}
+}
+
+func (eng *Engine) Transport() Transport {
+ return Transport{C.pn_connection_engine_transport(&eng.engine)}
+}
+
func (eng *Engine) String() string {
return fmt.Sprintf("%s-%s", eng.conn.LocalAddr(), eng.conn.RemoteAddr())
}
@@ -221,26 +211,25 @@ func (eng *Engine) InjectWait(f func() error) error {
// the incoming connnection such as use of SASL and SSL.
// Must be called before Run()
//
-func (eng *Engine) Server() { eng.transport.SetServer() }
+func (eng *Engine) Server() { eng.Transport().SetServer() }
-func (eng *Engine) disconnect() {
- eng.transport.CloseHead()
- eng.transport.CloseTail()
+// FIXME aconway 2016-06-21: rename
+func (eng *Engine) disconnect() { // FIXME aconway 2016-06-21: disconnected
eng.conn.Close()
- eng.dispatch()
+ C.pn_connection_engine_disconnected(&eng.engine)
}
// Close the engine's connection.
// If err != nil pass it to the remote end as the close condition.
// Returns when the remote end closes or disconnects.
func (eng *Engine) Close(err error) {
- eng.Inject(func() { CloseError(eng.connection, err) })
+ eng.Inject(func() { CloseError(eng.Connection(), err) })
<-eng.running
}
// CloseTimeout like Close but disconnect if the remote end doesn't close within timeout.
func (eng *Engine) CloseTimeout(err error, timeout time.Duration) {
- eng.Inject(func() { CloseError(eng.connection, err) })
+ eng.Inject(func() { CloseError(eng.Connection(), err) })
select {
case <-eng.running:
case <-time.After(timeout):
@@ -251,7 +240,7 @@ func (eng *Engine) CloseTimeout(err error, timeout time.Duration) {
// Disconnect the engine's connection immediately without an AMQP close.
// Process any termination events before returning.
func (eng *Engine) Disconnect(err error) {
- eng.Inject(func() { eng.transport.Condition().SetError(err); eng.disconnect() })
+ eng.Inject(func() { eng.Transport().Condition().SetError(err); eng.disconnect() })
<-eng.running
}
@@ -259,151 +248,121 @@ func (eng *Engine) Disconnect(err error) {
// disconnected. You can check for errors after exit with Engine.Error().
//
func (eng *Engine) Run() error {
+ // Channels for read and write buffers going in and out of the read/write goroutines.
+ // The channels are unbuffered: we want to exchange buffers in seuquence.
+ readsIn, writesIn := make(chan []byte), make(chan []byte)
+ readsOut, writesOut := make(chan []byte), make(chan []byte)
+
wait := sync.WaitGroup{}
wait.Add(2) // Read and write goroutines
- readErr := make(chan error, 1) // Don't block
- go func() { // Read goroutine
+ go func() { // Read goroutine
defer wait.Done()
for {
- rbuf := eng.read.buffer()
+ rbuf, ok := <-readsIn
+ if !ok {
+ return
+ }
n, err := eng.conn.Read(rbuf)
if n > 0 {
- eng.read.buffers <- rbuf[:n]
- }
- if err != nil {
- readErr <- err
- close(readErr)
- close(eng.read.buffers)
+ readsOut <- rbuf[:n]
+ } else if err != nil {
+ eng.inject <- func() {
+ eng.Transport().Condition().SetError(err)
+ C.pn_connection_engine_read_close(&eng.engine)
+ }
return
}
}
}()
- writeErr := make(chan error, 1) // Don't block
- go func() { // Write goroutine
+ go func() { // Write goroutine
defer wait.Done()
for {
- wbuf, ok := <-eng.write.buffers
+ wbuf, ok := <-writesIn
if !ok {
return
}
- _, err := eng.conn.Write(wbuf)
- if err != nil {
- writeErr <- err
- close(writeErr)
+ n, err := eng.conn.Write(wbuf)
+ if n > 0 {
+ writesOut <- wbuf[:n]
+ } else if err != nil {
+ eng.inject <- func() {
+ eng.Transport().Condition().SetError(err)
+ C.pn_connection_engine_write_close(&eng.engine)
+ }
return
}
}
}()
- wbuf := eng.write.buffer()[:0]
-
- for !eng.transport.Closed() {
- if len(wbuf) == 0 {
- eng.pop(&wbuf)
+ for !C.pn_connection_engine_finished(&eng.engine) {
+ // Enable readIn/writeIn channles only if we have a buffer.
+ readBuf, writeBuf := eng.buffers()
+ var sendReads, sendWrites chan []byte
+ if readBuf != nil {
+ sendReads = readsIn
}
- // Don't set wchan unless there is something to write.
- var wchan chan []byte
- if len(wbuf) > 0 {
- wchan = eng.write.buffers
+ if writeBuf != nil {
+ sendWrites = writesIn
}
+ // Send buffers to the read/write goroutines if we have them.
+ // Get buffers from the read/write goroutines and process them
select {
- case buf, ok := <-eng.read.buffers: // Read a buffer
- if ok {
- eng.push(buf)
+
+ case sendReads <- readBuf:
+
+ case sendWrites <- writeBuf:
+
+ case buf := <-readsOut:
+ if len(buf) > 0 {
+ C.pn_connection_engine_read_done(&eng.engine, C.size_t(len(buf)))
+ } else {
+ panic(fmt.Sprintf("read buf %v", buf))
}
- case wchan <- wbuf: // Write a buffer
- wbuf = eng.write.buffer()[:0]
+
+ case buf := <-writesOut:
+ if len(buf) > 0 {
+ C.pn_connection_engine_write_done(&eng.engine, C.size_t(len(buf)))
+ } else {
+ panic(fmt.Sprintf("write buf %v", buf))
+ }
+
case f, ok := <-eng.inject: // Function injected from another goroutine
if ok {
f()
}
- case err := <-readErr:
- eng.transport.Condition().SetError(err)
- eng.transport.CloseTail()
- case err := <-writeErr:
- eng.transport.Condition().SetError(err)
- eng.transport.CloseHead()
}
- eng.dispatch()
- if eng.connection.State().RemoteClosed() && eng.connection.State().LocalClosed() {
- eng.disconnect()
+
+ for {
+ cevent := C.pn_connection_engine_dispatch(&eng.engine)
+ if cevent == nil {
+ break
+ }
+ event := makeEvent(cevent, eng)
+ for _, h := range eng.handlers {
+ h.HandleEvent(event)
+ }
}
}
- eng.err.Set(EndpointError(eng.connection))
- eng.err.Set(eng.transport.Condition().Error())
- close(eng.write.buffers)
+ eng.err.Set(EndpointError(eng.Connection()))
+ eng.err.Set(eng.Transport().Condition().Error())
+ close(readsIn)
+ close(writesIn)
eng.conn.Close() // Make sure connection is closed
- wait.Wait()
+ wait.Wait() // Wait for goroutines
+
close(eng.running) // Signal goroutines have exited and Error is set.
- if !eng.connection.IsNil() {
- eng.connection.Free()
- }
- if !eng.transport.IsNil() {
- eng.transport.Free()
- }
- if eng.collector != nil {
- C.pn_collector_free(eng.collector)
- }
+ C.pn_connection_engine_final(&eng.engine)
+
for _, h := range eng.handlers {
switch h := h.(type) {
case cHandler:
C.pn_handler_free(h.pn)
}
}
+ // FIXME aconway 2016-06-21: consistent error handling
return eng.err.Get()
}
-
-func minInt(a, b int) int {
- if a < b {
- return a
- } else {
- return b
- }
-}
-
-func (eng *Engine) pop(buf *[]byte) {
- pending := int(eng.transport.Pending())
- switch {
- case pending == int(C.PN_EOS):
- *buf = (*buf)[:]
- return
- case pending < 0:
- panic(fmt.Errorf("%s", PnErrorCode(pending)))
- }
- size := minInt(pending, cap(*buf))
- *buf = (*buf)[:size]
- if size == 0 {
- return
- }
- C.memcpy(unsafe.Pointer(&(*buf)[0]), eng.transport.Head(), C.size_t(size))
- assert(size > 0)
- eng.transport.Pop(uint(size))
-}
-
-func (eng *Engine) push(buf []byte) {
- buf2 := buf
- for len(buf2) > 0 {
- n := eng.transport.Push(buf2)
- if n <= 0 {
- panic(fmt.Errorf("error in transport: %s", PnErrorCode(n)))
- }
- buf2 = buf2[n:]
- }
-}
-
-func (eng *Engine) peek() *C.pn_event_t { return C.pn_collector_peek(eng.collector) }
-
-func (eng *Engine) dispatch() {
- for ce := eng.peek(); ce != nil; ce = eng.peek() {
- for _, h := range eng.handlers {
- h.HandleEvent(makeEvent(ce, eng))
- }
- C.pn_collector_pop(eng.collector)
- }
-}
-
-func (eng *Engine) Connection() Connection { return eng.connection }
http://git-wip-us.apache.org/repos/asf/qpid-proton/blob/c69cd100/proton-c/bindings/go/src/qpid.apache.org/proton/handlers.go
----------------------------------------------------------------------
diff --git a/proton-c/bindings/go/src/qpid.apache.org/proton/handlers.go b/proton-c/bindings/go/src/qpid.apache.org/proton/handlers.go
index 87c43ae..2a96d81 100644
--- a/proton-c/bindings/go/src/qpid.apache.org/proton/handlers.go
+++ b/proton-c/bindings/go/src/qpid.apache.org/proton/handlers.go
@@ -31,6 +31,8 @@ type EventHandler interface {
HandleEvent(e Event)
}
+// FIXME aconway 2016-06-21: get rid of C handlers?
+
// cHandler wraps a C pn_handler_t
type cHandler struct {
pn *C.pn_handler_t
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@qpid.apache.org
For additional commands, e-mail: commits-help@qpid.apache.org