| 115 | } |
| 116 | |
| 117 | int64_t MessageEndpoint::Write(uint64_t id, uint16_t size, uint64_t data){ |
| 118 | if(!peer.get()){ |
| 119 | return -ENOTCONN; |
| 120 | } |
| 121 | |
| 122 | if(size > maxMessageSize){ |
| 123 | return -EINVAL; |
| 124 | } |
| 125 | |
| 126 | Message msg; |
| 127 | |
| 128 | msg.id = id; |
| 129 | msg.size = size; |
| 130 | |
| 131 | if(msg.size <= 8){ |
| 132 | msg.data = data; |
| 133 | } else { |
| 134 | assert(data); |
| 135 | |
| 136 | msg.dataP = new uint8_t[maxMessageSize]; |
| 137 | |
| 138 | memcpy(msg.dataP, reinterpret_cast<uint8_t*>(data), size); |
| 139 | } |
| 140 | |
| 141 | acquireLock(&peer->waitingResponseLock); |
| 142 | for(auto it = peer->waitingResponse.begin(); it != peer->waitingResponse.end(); it++){ |
| 143 | if(it->item2.id == id){ |
| 144 | it->item1->Signal(); |
| 145 | *it->item2.ret = msg; |
| 146 | |
| 147 | if(debugLevelMessageEndpoint >= DebugLevelVerbose){ |
| 148 | Log::Info("[MessageEndpoint] Sending response (ID: %u, Size: %u) to peer", msg.id, msg.size); |
| 149 | } |
| 150 | |
| 151 | peer->waitingResponse.remove(it); |
| 152 | releaseLock(&peer->waitingResponseLock); |
| 153 | return 0; // Skip queue entirely |
| 154 | } |
| 155 | } |
| 156 | releaseLock(&peer->waitingResponseLock); |
| 157 | |
| 158 | queueAvailablilitySemaphore.Wait(); |
| 159 | |
| 160 | acquireLock(&peer->queueLock); |
| 161 | peer->queue.Enqueue(&msg); |
| 162 | |
| 163 | acquireLock(&peer->waitingLock); |
| 164 | while(peer->waiting.get_length() > 0){ |
| 165 | peer->waiting.remove_at(0)->Signal(); |
| 166 | } |
| 167 | releaseLock(&peer->waitingLock); |
| 168 | |
| 169 | if(debugLevelMessageEndpoint >= DebugLevelVerbose){ |
| 170 | Log::Info("[MessageEndpoint] Sending message (ID: %u, Size: %u) to peer", msg.id, msg.size); |
| 171 | } |
| 172 | |
| 173 | releaseLock(&peer->queueLock); |
| 174 | return 0; |