diff options
Diffstat (limited to 'go')
| -rw-r--r-- | go/nxt_cgo_lib.c | 20 | ||||
| -rw-r--r-- | go/observable.go | 32 | ||||
| -rw-r--r-- | go/response.go | 36 |
3 files changed, 77 insertions, 11 deletions
diff --git a/go/nxt_cgo_lib.c b/go/nxt_cgo_lib.c index 5cb31b5a..a4fef9ea 100644 --- a/go/nxt_cgo_lib.c +++ b/go/nxt_cgo_lib.c @@ -19,6 +19,7 @@ static ssize_t nxt_cgo_port_send(nxt_unit_ctx_t *, nxt_unit_port_id_t *port_id, const void *buf, size_t buf_size, const void *oob, size_t oob_size); static ssize_t nxt_cgo_port_recv(nxt_unit_ctx_t *, nxt_unit_port_id_t *port_id, void *buf, size_t buf_size, void *oob, size_t oob_size); +static void nxt_cgo_shm_ack_handler(nxt_unit_ctx_t *ctx); int nxt_cgo_run(uintptr_t handler) @@ -34,6 +35,7 @@ nxt_cgo_run(uintptr_t handler) init.callbacks.remove_port = nxt_cgo_remove_port; init.callbacks.port_send = nxt_cgo_port_send; init.callbacks.port_recv = nxt_cgo_port_recv; + init.callbacks.shm_ack_handler = nxt_cgo_shm_ack_handler; init.data = (void *) handler; @@ -137,6 +139,13 @@ nxt_cgo_port_recv(nxt_unit_ctx_t *ctx, nxt_unit_port_id_t *port_id, } +static void +nxt_cgo_shm_ack_handler(nxt_unit_ctx_t *ctx) +{ + return nxt_go_shm_ack_handler(); +} + + int nxt_cgo_response_create(uintptr_t req, int status, int fields, uint32_t fields_size) @@ -166,15 +175,8 @@ nxt_cgo_response_send(uintptr_t req) ssize_t nxt_cgo_response_write(uintptr_t req, uintptr_t start, uint32_t len) { - int rc; - - rc = nxt_unit_response_write((nxt_unit_request_info_t *) req, - (void *) start, len); - if (rc != NXT_UNIT_OK) { - return -1; - } - - return len; + return nxt_unit_response_write_nb((nxt_unit_request_info_t *) req, + (void *) start, len, 0); } diff --git a/go/observable.go b/go/observable.go new file mode 100644 index 00000000..9a38802c --- /dev/null +++ b/go/observable.go @@ -0,0 +1,32 @@ +/* + * Copyright (C) NGINX, Inc. + */ + +package unit + +import ( + "sync" +) + +type observable struct { + sync.Mutex + observers []chan int +} + +func (o *observable) attach(c chan int) { + o.Lock() + defer o.Unlock() + + o.observers = append(o.observers, c) +} + +func (o *observable) notify(e int) { + o.Lock() + defer o.Unlock() + + for _, v := range o.observers { + v <- e + } + + o.observers = nil +} diff --git a/go/response.go b/go/response.go index 767d66b7..bfa79656 100644 --- a/go/response.go +++ b/go/response.go @@ -19,6 +19,7 @@ type response struct { headerSent bool req *http.Request c_req C.uintptr_t + ch chan int } func new_response(c_req C.uintptr_t, req *http.Request) *response { @@ -40,8 +41,26 @@ func (r *response) Write(p []byte) (n int, err error) { r.WriteHeader(http.StatusOK) } - res := C.nxt_cgo_response_write(r.c_req, buf_ref(p), C.uint32_t(len(p))) - return int(res), nil + l := len(p) + written := int(0) + br := buf_ref(p) + + for written < l { + res := C.nxt_cgo_response_write(r.c_req, br, C.uint32_t(l - written)) + + written += int(res) + br += C.uintptr_t(res) + + if (written < l) { + if r.ch == nil { + r.ch = make(chan int, 2) + } + + wait_shm_ack(r.ch) + } + } + + return written, nil } func (r *response) WriteHeader(code int) { @@ -85,3 +104,16 @@ func (r *response) Flush() { r.WriteHeader(http.StatusOK) } } + +var observer_registry_ observable + +func wait_shm_ack(c chan int) { + observer_registry_.attach(c) + + _ = <-c +} + +//export nxt_go_shm_ack_handler +func nxt_go_shm_ack_handler() { + observer_registry_.notify(1) +} |