| 22 | } |
| 23 | |
| 24 | bool scribe::consume(scheduler* ctx, const void*, size_t num_bytes) { |
| 25 | CAF_ASSERT(ctx != nullptr); |
| 26 | auto lg = log::io::trace("num_bytes = {}", num_bytes); |
| 27 | if (detached()) |
| 28 | // we are already disconnected from the broker while the multiplexer |
| 29 | // did not yet remove the socket, this can happen if an I/O event causes |
| 30 | // the broker to call close_all() while the pollset contained |
| 31 | // further activities for the broker |
| 32 | return false; |
| 33 | // keep a strong reference to our parent until we leave scope |
| 34 | // to avoid UB when becoming detached during invocation |
| 35 | auto guard = parent_; |
| 36 | auto& buf = rd_buf(); |
| 37 | CAF_ASSERT(buf.size() >= num_bytes); |
| 38 | // make sure size is correct, swap into message, and then call client |
| 39 | buf.resize(num_bytes); |
| 40 | auto& msg_buf = msg().buf; |
| 41 | msg_buf.swap(buf); |
| 42 | auto result = invoke_mailbox_element(ctx); |
| 43 | // swap buffer back to stream and implicitly flush wr_buf() |
| 44 | msg_buf.swap(buf); |
| 45 | flush(); |
| 46 | return result; |
| 47 | } |
| 48 | |
| 49 | void scribe::data_transferred(scheduler* ctx, size_t written, |
| 50 | size_t remaining) { |