1 /* 2 * Copyright (C) 2007 The Android Open Source Project 3 * 4 * Licensed under the Apache License, Version 2.0 (the "License"); 5 * you may not use this file except in compliance with the License. 6 * You may obtain a copy of the License at 7 * 8 * http://www.apache.org/licenses/LICENSE-2.0 9 * 10 * Unless required by applicable law or agreed to in writing, software 11 * distributed under the License is distributed on an "AS IS" BASIS, 12 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 13 * See the License for the specific language governing permissions and 14 * limitations under the License. 15 */ 16 17 #define TRACE_TAG SOCKETS 18 19 #include "sysdeps.h" 20 21 #include <ctype.h> 22 #include <errno.h> 23 #include <stdio.h> 24 #include <stdlib.h> 25 #include <string.h> 26 #include <unistd.h> 27 28 #include <algorithm> 29 #include <mutex> 30 #include <string> 31 #include <vector> 32 33 #if !ADB_HOST 34 #include <android-base/properties.h> 35 #include <log/log_properties.h> 36 #endif 37 38 #include "adb.h" 39 #include "adb_io.h" 40 #include "range.h" 41 #include "transport.h" 42 43 static std::recursive_mutex& local_socket_list_lock = *new std::recursive_mutex(); 44 static unsigned local_socket_next_id = 1; 45 46 static auto& local_socket_list = *new std::vector<asocket*>(); 47 48 /* the the list of currently closing local sockets. 49 ** these have no peer anymore, but still packets to 50 ** write to their fd. 51 */ 52 static auto& local_socket_closing_list = *new std::vector<asocket*>(); 53 54 // Parse the global list of sockets to find one with id |local_id|. 55 // If |peer_id| is not 0, also check that it is connected to a peer 56 // with id |peer_id|. Returns an asocket handle on success, NULL on failure. 57 asocket* find_local_socket(unsigned local_id, unsigned peer_id) { 58 asocket* result = nullptr; 59 60 std::lock_guard<std::recursive_mutex> lock(local_socket_list_lock); 61 for (asocket* s : local_socket_list) { 62 if (s->id != local_id) { 63 continue; 64 } 65 if (peer_id == 0 || (s->peer && s->peer->id == peer_id)) { 66 result = s; 67 } 68 break; 69 } 70 71 return result; 72 } 73 74 void install_local_socket(asocket* s) { 75 std::lock_guard<std::recursive_mutex> lock(local_socket_list_lock); 76 77 s->id = local_socket_next_id++; 78 79 // Socket ids should never be 0. 80 if (local_socket_next_id == 0) { 81 fatal("local socket id overflow"); 82 } 83 84 local_socket_list.push_back(s); 85 } 86 87 void remove_socket(asocket* s) { 88 std::lock_guard<std::recursive_mutex> lock(local_socket_list_lock); 89 for (auto list : { &local_socket_list, &local_socket_closing_list }) { 90 list->erase(std::remove_if(list->begin(), list->end(), [s](asocket* x) { return x == s; }), 91 list->end()); 92 } 93 } 94 95 void close_all_sockets(atransport* t) { 96 /* this is a little gross, but since s->close() *will* modify 97 ** the list out from under you, your options are limited. 98 */ 99 std::lock_guard<std::recursive_mutex> lock(local_socket_list_lock); 100 restart: 101 for (asocket* s : local_socket_list) { 102 if (s->transport == t || (s->peer && s->peer->transport == t)) { 103 s->close(s); 104 goto restart; 105 } 106 } 107 } 108 109 enum class SocketFlushResult { 110 Destroyed, 111 TryAgain, 112 Completed, 113 }; 114 115 static SocketFlushResult local_socket_flush_incoming(asocket* s) { 116 while (!s->packet_queue.empty()) { 117 Range& r = s->packet_queue.front(); 118 119 int rc = adb_write(s->fd, r.data(), r.size()); 120 if (rc == static_cast<int>(r.size())) { 121 s->packet_queue.pop_front(); 122 } else if (rc > 0) { 123 r.drop_front(rc); 124 fdevent_add(&s->fde, FDE_WRITE); 125 return SocketFlushResult::TryAgain; 126 } else if (rc == -1 && errno == EAGAIN) { 127 fdevent_add(&s->fde, FDE_WRITE); 128 return SocketFlushResult::TryAgain; 129 } 130 131 // We failed to write, but it's possible that we can still read from the socket. 132 // Give that a try before giving up. 133 s->has_write_error = true; 134 break; 135 } 136 137 // If we sent the last packet of a closing socket, we can now destroy it. 138 if (s->closing) { 139 s->close(s); 140 return SocketFlushResult::Destroyed; 141 } 142 143 fdevent_del(&s->fde, FDE_WRITE); 144 return SocketFlushResult::Completed; 145 } 146 147 // Returns false if the socket has been closed and destroyed as a side-effect of this function. 148 static bool local_socket_flush_outgoing(asocket* s) { 149 const size_t max_payload = s->get_max_payload(); 150 std::string data; 151 data.resize(max_payload); 152 char* x = &data[0]; 153 size_t avail = max_payload; 154 int r = 0; 155 int is_eof = 0; 156 157 while (avail > 0) { 158 r = adb_read(s->fd, x, avail); 159 D("LS(%d): post adb_read(fd=%d,...) r=%d (errno=%d) avail=%zu", s->id, s->fd, r, 160 r < 0 ? errno : 0, avail); 161 if (r == -1) { 162 if (errno == EAGAIN) { 163 break; 164 } 165 } else if (r > 0) { 166 avail -= r; 167 x += r; 168 continue; 169 } 170 171 /* r = 0 or unhandled error */ 172 is_eof = 1; 173 break; 174 } 175 D("LS(%d): fd=%d post avail loop. r=%d is_eof=%d forced_eof=%d", s->id, s->fd, r, is_eof, 176 s->fde.force_eof); 177 178 if (avail != max_payload && s->peer) { 179 data.resize(max_payload - avail); 180 181 // s->peer->enqueue() may call s->close() and free s, 182 // so save variables for debug printing below. 183 unsigned saved_id = s->id; 184 int saved_fd = s->fd; 185 r = s->peer->enqueue(s->peer, std::move(data)); 186 D("LS(%u): fd=%d post peer->enqueue(). r=%d", saved_id, saved_fd, r); 187 188 if (r < 0) { 189 // Error return means they closed us as a side-effect and we must 190 // return immediately. 191 // 192 // Note that if we still have buffered packets, the socket will be 193 // placed on the closing socket list. This handler function will be 194 // called again to process FDE_WRITE events. 195 return false; 196 } 197 198 if (r > 0) { 199 /* if the remote cannot accept further events, 200 ** we disable notification of READs. They'll 201 ** be enabled again when we get a call to ready() 202 */ 203 fdevent_del(&s->fde, FDE_READ); 204 } 205 } 206 207 // Don't allow a forced eof if data is still there. 208 if ((s->fde.force_eof && !r) || is_eof) { 209 D(" closing because is_eof=%d r=%d s->fde.force_eof=%d", is_eof, r, s->fde.force_eof); 210 s->close(s); 211 return false; 212 } 213 214 return true; 215 } 216 217 static int local_socket_enqueue(asocket* s, std::string data) { 218 D("LS(%d): enqueue %zu", s->id, data.size()); 219 220 Range r(std::move(data)); 221 s->packet_queue.push_back(std::move(r)); 222 switch (local_socket_flush_incoming(s)) { 223 case SocketFlushResult::Destroyed: 224 return -1; 225 226 case SocketFlushResult::TryAgain: 227 return 1; 228 229 case SocketFlushResult::Completed: 230 return 0; 231 } 232 233 return !s->packet_queue.empty(); 234 } 235 236 static void local_socket_ready(asocket* s) { 237 /* far side is ready for data, pay attention to 238 readable events */ 239 fdevent_add(&s->fde, FDE_READ); 240 } 241 242 // be sure to hold the socket list lock when calling this 243 static void local_socket_destroy(asocket* s) { 244 int exit_on_close = s->exit_on_close; 245 246 D("LS(%d): destroying fde.fd=%d", s->id, s->fde.fd); 247 248 /* IMPORTANT: the remove closes the fd 249 ** that belongs to this socket 250 */ 251 fdevent_remove(&s->fde); 252 253 remove_socket(s); 254 delete s; 255 256 if (exit_on_close) { 257 D("local_socket_destroy: exiting"); 258 exit(1); 259 } 260 } 261 262 static void local_socket_close(asocket* s) { 263 D("entered local_socket_close. LS(%d) fd=%d", s->id, s->fd); 264 std::lock_guard<std::recursive_mutex> lock(local_socket_list_lock); 265 if (s->peer) { 266 D("LS(%d): closing peer. peer->id=%d peer->fd=%d", s->id, s->peer->id, s->peer->fd); 267 /* Note: it's important to call shutdown before disconnecting from 268 * the peer, this ensures that remote sockets can still get the id 269 * of the local socket they're connected to, to send a CLOSE() 270 * protocol event. */ 271 if (s->peer->shutdown) { 272 s->peer->shutdown(s->peer); 273 } 274 s->peer->peer = nullptr; 275 s->peer->close(s->peer); 276 s->peer = nullptr; 277 } 278 279 /* If we are already closing, or if there are no 280 ** pending packets, destroy immediately 281 */ 282 if (s->closing || s->has_write_error || s->packet_queue.empty()) { 283 int id = s->id; 284 local_socket_destroy(s); 285 D("LS(%d): closed", id); 286 return; 287 } 288 289 /* otherwise, put on the closing list 290 */ 291 D("LS(%d): closing", s->id); 292 s->closing = 1; 293 fdevent_del(&s->fde, FDE_READ); 294 remove_socket(s); 295 D("LS(%d): put on socket_closing_list fd=%d", s->id, s->fd); 296 local_socket_closing_list.push_back(s); 297 CHECK_EQ(FDE_WRITE, s->fde.state & FDE_WRITE); 298 } 299 300 static void local_socket_event_func(int fd, unsigned ev, void* _s) { 301 asocket* s = reinterpret_cast<asocket*>(_s); 302 D("LS(%d): event_func(fd=%d(==%d), ev=%04x)", s->id, s->fd, fd, ev); 303 304 /* put the FDE_WRITE processing before the FDE_READ 305 ** in order to simplify the code. 306 */ 307 if (ev & FDE_WRITE) { 308 switch (local_socket_flush_incoming(s)) { 309 case SocketFlushResult::Destroyed: 310 return; 311 312 case SocketFlushResult::TryAgain: 313 break; 314 315 case SocketFlushResult::Completed: 316 s->peer->ready(s->peer); 317 break; 318 } 319 } 320 321 if (ev & FDE_READ) { 322 if (!local_socket_flush_outgoing(s)) { 323 return; 324 } 325 } 326 327 if (ev & FDE_ERROR) { 328 /* this should be caught be the next read or write 329 ** catching it here means we may skip the last few 330 ** bytes of readable data. 331 */ 332 D("LS(%d): FDE_ERROR (fd=%d)", s->id, s->fd); 333 return; 334 } 335 } 336 337 asocket* create_local_socket(int fd) { 338 asocket* s = new asocket(); 339 s->fd = fd; 340 s->enqueue = local_socket_enqueue; 341 s->ready = local_socket_ready; 342 s->shutdown = NULL; 343 s->close = local_socket_close; 344 install_local_socket(s); 345 346 fdevent_install(&s->fde, fd, local_socket_event_func, s); 347 D("LS(%d): created (fd=%d)", s->id, s->fd); 348 return s; 349 } 350 351 asocket* create_local_service_socket(const char* name, atransport* transport) { 352 #if !ADB_HOST 353 if (!strcmp(name, "jdwp")) { 354 return create_jdwp_service_socket(); 355 } 356 if (!strcmp(name, "track-jdwp")) { 357 return create_jdwp_tracker_service_socket(); 358 } 359 #endif 360 int fd = service_to_fd(name, transport); 361 if (fd < 0) { 362 return nullptr; 363 } 364 365 asocket* s = create_local_socket(fd); 366 D("LS(%d): bound to '%s' via %d", s->id, name, fd); 367 368 #if !ADB_HOST 369 if ((!strncmp(name, "root:", 5) && getuid() != 0 && __android_log_is_debuggable()) || 370 (!strncmp(name, "unroot:", 7) && getuid() == 0) || 371 !strncmp(name, "usb:", 4) || 372 !strncmp(name, "tcpip:", 6)) { 373 D("LS(%d): enabling exit_on_close", s->id); 374 s->exit_on_close = 1; 375 } 376 #endif 377 378 return s; 379 } 380 381 #if ADB_HOST 382 static asocket* create_host_service_socket(const char* name, const char* serial, 383 TransportId transport_id) { 384 asocket* s; 385 386 s = host_service_to_socket(name, serial, transport_id); 387 388 if (s != NULL) { 389 D("LS(%d) bound to '%s'", s->id, name); 390 return s; 391 } 392 393 return s; 394 } 395 #endif /* ADB_HOST */ 396 397 static int remote_socket_enqueue(asocket* s, std::string data) { 398 D("entered remote_socket_enqueue RS(%d) WRITE fd=%d peer.fd=%d", s->id, s->fd, s->peer->fd); 399 apacket* p = get_apacket(); 400 401 p->msg.command = A_WRTE; 402 p->msg.arg0 = s->peer->id; 403 p->msg.arg1 = s->id; 404 405 if (data.size() > MAX_PAYLOAD) { 406 put_apacket(p); 407 return -1; 408 } 409 410 p->payload = std::move(data); 411 p->msg.data_length = p->payload.size(); 412 413 send_packet(p, s->transport); 414 return 1; 415 } 416 417 static void remote_socket_ready(asocket* s) { 418 D("entered remote_socket_ready RS(%d) OKAY fd=%d peer.fd=%d", s->id, s->fd, s->peer->fd); 419 apacket* p = get_apacket(); 420 p->msg.command = A_OKAY; 421 p->msg.arg0 = s->peer->id; 422 p->msg.arg1 = s->id; 423 send_packet(p, s->transport); 424 } 425 426 static void remote_socket_shutdown(asocket* s) { 427 D("entered remote_socket_shutdown RS(%d) CLOSE fd=%d peer->fd=%d", s->id, s->fd, 428 s->peer ? s->peer->fd : -1); 429 apacket* p = get_apacket(); 430 p->msg.command = A_CLSE; 431 if (s->peer) { 432 p->msg.arg0 = s->peer->id; 433 } 434 p->msg.arg1 = s->id; 435 send_packet(p, s->transport); 436 } 437 438 static void remote_socket_close(asocket* s) { 439 if (s->peer) { 440 s->peer->peer = 0; 441 D("RS(%d) peer->close()ing peer->id=%d peer->fd=%d", s->id, s->peer->id, s->peer->fd); 442 s->peer->close(s->peer); 443 } 444 D("entered remote_socket_close RS(%d) CLOSE fd=%d peer->fd=%d", s->id, s->fd, 445 s->peer ? s->peer->fd : -1); 446 D("RS(%d): closed", s->id); 447 delete s; 448 } 449 450 // Create a remote socket to exchange packets with a remote service through transport 451 // |t|. Where |id| is the socket id of the corresponding service on the other 452 // side of the transport (it is allocated by the remote side and _cannot_ be 0). 453 // Returns a new non-NULL asocket handle. 454 asocket* create_remote_socket(unsigned id, atransport* t) { 455 if (id == 0) { 456 fatal("invalid remote socket id (0)"); 457 } 458 asocket* s = new asocket(); 459 s->id = id; 460 s->enqueue = remote_socket_enqueue; 461 s->ready = remote_socket_ready; 462 s->shutdown = remote_socket_shutdown; 463 s->close = remote_socket_close; 464 s->transport = t; 465 466 D("RS(%d): created", s->id); 467 return s; 468 } 469 470 void connect_to_remote(asocket* s, const char* destination) { 471 D("Connect_to_remote call RS(%d) fd=%d", s->id, s->fd); 472 apacket* p = get_apacket(); 473 474 D("LS(%d): connect('%s')", s->id, destination); 475 p->msg.command = A_OPEN; 476 p->msg.arg0 = s->id; 477 478 // adbd expects a null-terminated string. 479 p->payload = destination; 480 p->payload.push_back('\0'); 481 p->msg.data_length = p->payload.size(); 482 483 if (p->msg.data_length > s->get_max_payload()) { 484 fatal("destination oversized"); 485 } 486 487 send_packet(p, s->transport); 488 } 489 490 /* this is used by magic sockets to rig local sockets to 491 send the go-ahead message when they connect */ 492 static void local_socket_ready_notify(asocket* s) { 493 s->ready = local_socket_ready; 494 s->shutdown = NULL; 495 s->close = local_socket_close; 496 SendOkay(s->fd); 497 s->ready(s); 498 } 499 500 /* this is used by magic sockets to rig local sockets to 501 send the failure message if they are closed before 502 connected (to avoid closing them without a status message) */ 503 static void local_socket_close_notify(asocket* s) { 504 s->ready = local_socket_ready; 505 s->shutdown = NULL; 506 s->close = local_socket_close; 507 SendFail(s->fd, "closed"); 508 s->close(s); 509 } 510 511 static unsigned unhex(const char* s, int len) { 512 unsigned n = 0, c; 513 514 while (len-- > 0) { 515 switch ((c = *s++)) { 516 case '0': 517 case '1': 518 case '2': 519 case '3': 520 case '4': 521 case '5': 522 case '6': 523 case '7': 524 case '8': 525 case '9': 526 c -= '0'; 527 break; 528 case 'a': 529 case 'b': 530 case 'c': 531 case 'd': 532 case 'e': 533 case 'f': 534 c = c - 'a' + 10; 535 break; 536 case 'A': 537 case 'B': 538 case 'C': 539 case 'D': 540 case 'E': 541 case 'F': 542 c = c - 'A' + 10; 543 break; 544 default: 545 return 0xffffffff; 546 } 547 548 n = (n << 4) | c; 549 } 550 551 return n; 552 } 553 554 #if ADB_HOST 555 556 namespace internal { 557 558 // Returns the position in |service| following the target serial parameter. Serial format can be 559 // any of: 560 // * [tcp:|udp:]<serial>[:<port>]:<command> 561 // * <prefix>:<serial>:<command> 562 // Where <port> must be a base-10 number and <prefix> may be any of {usb,product,model,device}. 563 // 564 // The returned pointer will point to the ':' just before <command>, or nullptr if not found. 565 char* skip_host_serial(char* service) { 566 static const std::vector<std::string>& prefixes = 567 *(new std::vector<std::string>{"usb:", "product:", "model:", "device:"}); 568 569 for (const std::string& prefix : prefixes) { 570 if (!strncmp(service, prefix.c_str(), prefix.length())) { 571 return strchr(service + prefix.length(), ':'); 572 } 573 } 574 575 // For fastboot compatibility, ignore protocol prefixes. 576 if (!strncmp(service, "tcp:", 4) || !strncmp(service, "udp:", 4)) { 577 service += 4; 578 } 579 580 // Check for an IPv6 address. `adb connect` creates the serial number from the canonical 581 // network address so it will always have the [] delimiters. 582 if (service[0] == '[') { 583 char* ipv6_end = strchr(service, ']'); 584 if (ipv6_end != nullptr) { 585 service = ipv6_end; 586 } 587 } 588 589 // The next colon we find must either begin the port field or the command field. 590 char* colon_ptr = strchr(service, ':'); 591 if (!colon_ptr) { 592 // No colon in service string. 593 return nullptr; 594 } 595 596 // If the next field is only decimal digits and ends with another colon, it's a port. 597 char* serial_end = colon_ptr; 598 if (isdigit(serial_end[1])) { 599 serial_end++; 600 while (*serial_end && isdigit(*serial_end)) { 601 serial_end++; 602 } 603 if (*serial_end != ':') { 604 // Something other than "<port>:" was found, this must be the command field instead. 605 serial_end = colon_ptr; 606 } 607 } 608 return serial_end; 609 } 610 611 } // namespace internal 612 613 #endif // ADB_HOST 614 615 static int smart_socket_enqueue(asocket* s, std::string data) { 616 #if ADB_HOST 617 char* service = nullptr; 618 char* serial = nullptr; 619 TransportId transport_id = 0; 620 TransportType type = kTransportAny; 621 #endif 622 623 D("SS(%d): enqueue %zu", s->id, data.size()); 624 625 if (s->smart_socket_data.empty()) { 626 s->smart_socket_data = std::move(data); 627 } else { 628 std::copy(data.begin(), data.end(), std::back_inserter(s->smart_socket_data)); 629 } 630 631 /* don't bother if we can't decode the length */ 632 if (s->smart_socket_data.size() < 4) { 633 return 0; 634 } 635 636 uint32_t len = unhex(s->smart_socket_data.data(), 4); 637 if (len == 0 || len > MAX_PAYLOAD) { 638 D("SS(%d): bad size (%u)", s->id, len); 639 goto fail; 640 } 641 642 D("SS(%d): len is %u", s->id, len); 643 /* can't do anything until we have the full header */ 644 if ((len + 4) > s->smart_socket_data.size()) { 645 D("SS(%d): waiting for %zu more bytes", s->id, len + 4 - s->smart_socket_data.size()); 646 return 0; 647 } 648 649 s->smart_socket_data[len + 4] = 0; 650 651 D("SS(%d): '%s'", s->id, (char*)(s->smart_socket_data.data() + 4)); 652 653 #if ADB_HOST 654 service = &s->smart_socket_data[4]; 655 if (!strncmp(service, "host-serial:", strlen("host-serial:"))) { 656 char* serial_end; 657 service += strlen("host-serial:"); 658 659 // serial number should follow "host:" and could be a host:port string. 660 serial_end = internal::skip_host_serial(service); 661 if (serial_end) { 662 *serial_end = 0; // terminate string 663 serial = service; 664 service = serial_end + 1; 665 } 666 } else if (!strncmp(service, "host-transport-id:", strlen("host-transport-id:"))) { 667 service += strlen("host-transport-id:"); 668 transport_id = strtoll(service, &service, 10); 669 670 if (*service != ':') { 671 return -1; 672 } 673 service++; 674 } else if (!strncmp(service, "host-usb:", strlen("host-usb:"))) { 675 type = kTransportUsb; 676 service += strlen("host-usb:"); 677 } else if (!strncmp(service, "host-local:", strlen("host-local:"))) { 678 type = kTransportLocal; 679 service += strlen("host-local:"); 680 } else if (!strncmp(service, "host:", strlen("host:"))) { 681 type = kTransportAny; 682 service += strlen("host:"); 683 } else { 684 service = nullptr; 685 } 686 687 if (service) { 688 asocket* s2; 689 690 /* some requests are handled immediately -- in that 691 ** case the handle_host_request() routine has sent 692 ** the OKAY or FAIL message and all we have to do 693 ** is clean up. 694 */ 695 if (handle_host_request(service, type, serial, transport_id, s->peer->fd, s) == 0) { 696 /* XXX fail message? */ 697 D("SS(%d): handled host service '%s'", s->id, service); 698 goto fail; 699 } 700 if (!strncmp(service, "transport", strlen("transport"))) { 701 D("SS(%d): okay transport", s->id); 702 s->smart_socket_data.clear(); 703 return 0; 704 } 705 706 /* try to find a local service with this name. 707 ** if no such service exists, we'll fail out 708 ** and tear down here. 709 */ 710 s2 = create_host_service_socket(service, serial, transport_id); 711 if (s2 == 0) { 712 D("SS(%d): couldn't create host service '%s'", s->id, service); 713 SendFail(s->peer->fd, "unknown host service"); 714 goto fail; 715 } 716 717 /* we've connected to a local host service, 718 ** so we make our peer back into a regular 719 ** local socket and bind it to the new local 720 ** service socket, acknowledge the successful 721 ** connection, and close this smart socket now 722 ** that its work is done. 723 */ 724 SendOkay(s->peer->fd); 725 726 s->peer->ready = local_socket_ready; 727 s->peer->shutdown = nullptr; 728 s->peer->close = local_socket_close; 729 s->peer->peer = s2; 730 s2->peer = s->peer; 731 s->peer = 0; 732 D("SS(%d): okay", s->id); 733 s->close(s); 734 735 /* initial state is "ready" */ 736 s2->ready(s2); 737 return 0; 738 } 739 #else /* !ADB_HOST */ 740 if (s->transport == nullptr) { 741 std::string error_msg = "unknown failure"; 742 s->transport = acquire_one_transport(kTransportAny, nullptr, 0, nullptr, &error_msg); 743 if (s->transport == nullptr) { 744 SendFail(s->peer->fd, error_msg); 745 goto fail; 746 } 747 } 748 #endif 749 750 if (!s->transport) { 751 SendFail(s->peer->fd, "device offline (no transport)"); 752 goto fail; 753 } else if (s->transport->GetConnectionState() == kCsOffline) { 754 /* if there's no remote we fail the connection 755 ** right here and terminate it 756 */ 757 SendFail(s->peer->fd, "device offline (transport offline)"); 758 goto fail; 759 } 760 761 /* instrument our peer to pass the success or fail 762 ** message back once it connects or closes, then 763 ** detach from it, request the connection, and 764 ** tear down 765 */ 766 s->peer->ready = local_socket_ready_notify; 767 s->peer->shutdown = nullptr; 768 s->peer->close = local_socket_close_notify; 769 s->peer->peer = 0; 770 /* give him our transport and upref it */ 771 s->peer->transport = s->transport; 772 773 connect_to_remote(s->peer, s->smart_socket_data.data() + 4); 774 s->peer = 0; 775 s->close(s); 776 return 1; 777 778 fail: 779 /* we're going to close our peer as a side-effect, so 780 ** return -1 to signal that state to the local socket 781 ** who is enqueueing against us 782 */ 783 s->close(s); 784 return -1; 785 } 786 787 static void smart_socket_ready(asocket* s) { 788 D("SS(%d): ready", s->id); 789 } 790 791 static void smart_socket_close(asocket* s) { 792 D("SS(%d): closed", s->id); 793 if (s->peer) { 794 s->peer->peer = 0; 795 s->peer->close(s->peer); 796 s->peer = 0; 797 } 798 delete s; 799 } 800 801 static asocket* create_smart_socket(void) { 802 D("Creating smart socket"); 803 asocket* s = new asocket(); 804 s->enqueue = smart_socket_enqueue; 805 s->ready = smart_socket_ready; 806 s->shutdown = NULL; 807 s->close = smart_socket_close; 808 809 D("SS(%d)", s->id); 810 return s; 811 } 812 813 void connect_to_smartsocket(asocket* s) { 814 D("Connecting to smart socket"); 815 asocket* ss = create_smart_socket(); 816 s->peer = ss; 817 ss->peer = s; 818 s->ready(s); 819 } 820 821 size_t asocket::get_max_payload() const { 822 size_t max_payload = MAX_PAYLOAD; 823 if (transport) { 824 max_payload = std::min(max_payload, transport->get_max_payload()); 825 } 826 if (peer && peer->transport) { 827 max_payload = std::min(max_payload, peer->transport->get_max_payload()); 828 } 829 return max_payload; 830 } 831
