// TODO: BUG: The Peak on Windows rarely doesn't close the input thread.
//   This causes the UI thread to hang on exit.
void *input_thread(void *_t, struct dll_io_bridge * _io_bridge)
{
    i_bridge = _io_bridge;

    atransport *t = (atransport *)_t;
    apacket *p;
    int active = 0;

    D("%s: starting transport input thread, reading from fd %d\n",
       t->serial, t->fd);

    for(;;){
        if(read_packet(t->fd, t->serial, &p)) {
            D("%s: failed to read apacket from transport on fd %d\n",
               t->serial, t->fd );
            break;
        }
        if(p->msg.command == A_SYNC){
            if(p->msg.arg0 == 0) {
                D("%s: transport SYNC offline\n", t->serial);
                put_apacket(p);
                break;
            } else {
                if(p->msg.arg1 == t->sync_token) {
                    D("%s: transport SYNC online\n", t->serial);
                    active = 1;
                } else {
                    D("%s: transport ignoring SYNC %d != %d\n",
                      t->serial, p->msg.arg1, t->sync_token);
                }
            }
        } else {
            if(active) {
                D("%s: transport got packet %d, sending to remote\n", t->serial, p->msg.command);
                t->write_to_remote(p, t);
            } else {
                D("%s: transport ignoring packet while offline\n", t->serial);
            }
        }

        put_apacket(p);
    }

    // this is necessary to avoid a race condition that occured when a transport closes
    // while a client socket is still active.
	D("Pre-close sockets input-thread\n");
    close_all_sockets(t);

    D("%s: transport input thread is exiting, fd %d\n", t->serial, t->fd);
#ifdef WIN32
    kick_transport(t, i_bridge->AdbCloseHandle);
#else
    kick_transport(t, NULL);
#endif
	D("Post-kick transport input-thread\n");
    transport_unref(t);
	D("Post-unref transport input-thread\n");

    return 0;
}

static void *output_thread(void *_t)
{
    atransport *t = reinterpret_cast<atransport*>(_t);
    apacket *p;

    ADB_LOGD(ADB_TSPT,
             "%s: starting transport output thread on fd %d, SYNC online (%d)",
             t->serial, t->fd, t->sync_token + 1);
    p = get_apacket();
    p->msg.command = A_SYNC;
    p->msg.arg0 = 1;
    p->msg.arg1 = ++(t->sync_token);
    p->msg.magic = A_SYNC ^ 0xffffffff;
    if (write_packet(t->fd, t->serial, &p)) {
        put_apacket(p);
        ADB_LOGE(ADB_TSPT, "%s: failed to write SYNC packet", t->serial);
        goto oops;
    }

    ADB_LOGD(ADB_TSPT, "%s: data pump started", t->serial);
    for (;;) {
        p = get_apacket();

        if (t->read_from_remote(p, t) == 0) {
            ADB_LOGD(ADB_TSPT,
                     "%s: received remote packet, sending to transport",
                     t->serial);
            if (write_packet(t->fd, t->serial, &p)) {
                put_apacket(p);
                ADB_LOGE(ADB_TSPT,
                         "%s: failed to write apacket to transport", t->serial);
                goto oops;
            }
        } else {
            ADB_LOGE(ADB_TSPT,
                     "%s: remote read failed for transport", t->serial);
            put_apacket(p);
            break;
        }
    }

    ADB_LOGD(ADB_TSPT, "%s: SYNC offline for transport", t->serial);
    p = get_apacket();
    p->msg.command = A_SYNC;
    p->msg.arg0 = 0;
    p->msg.arg1 = 0;
    p->msg.magic = A_SYNC ^ 0xffffffff;
    if (write_packet(t->fd, t->serial, &p)) {
        put_apacket(p);
        ADB_LOGW(ADB_TSPT,
                 "%s: failed to write SYNC apacket to transport", t->serial);
    }

oops:
    ADB_LOGD(ADB_TSPT, "%s: transport output thread is exiting", t->serial);
    kick_transport(t);
    transport_unref(t);
    return 0;
}

// The transport is opened by transport_register_func before
// the read_transport and write_transport threads are started.
//
// The read_transport thread issues a SYNC(1, token) message to let
// the write_transport thread know to start things up.  In the event
// of transport IO failure, the read_transport thread will post a
// SYNC(0,0) message to ensure shutdown.
//
// The transport will not actually be closed until both threads exit, but the threads
// will kick the transport on their way out to disconnect the underlying device.
//
// read_transport thread reads data from a transport (representing a usb/tcp connection),
// and makes the main thread call handle_packet().
static void *read_transport_thread(void *_t)
{
    atransport *t = reinterpret_cast<atransport*>(_t);
    apacket *p;

    adb_thread_setname(android::base::StringPrintf("<-%s",
                                                   (t->serial != nullptr ? t->serial : "transport")));
    D("%s: starting read_transport thread on fd %d, SYNC online (%d)",
       t->serial, t->fd, t->sync_token + 1);
    p = get_apacket();
    p->msg.command = A_SYNC;
    p->msg.arg0 = 1;
    p->msg.arg1 = ++(t->sync_token);
    p->msg.magic = A_SYNC ^ 0xffffffff;
    if(write_packet(t->fd, t->serial, &p)) {
        put_apacket(p);
        D("%s: failed to write SYNC packet", t->serial);
        goto oops;
    }

    D("%s: data pump started", t->serial);
    for(;;) {
        p = get_apacket();

        if(t->read_from_remote(p, t) == 0){
            D("%s: received remote packet, sending to transport",
              t->serial);
            if(write_packet(t->fd, t->serial, &p)){
                put_apacket(p);
                D("%s: failed to write apacket to transport", t->serial);
                goto oops;
            }
        } else {
            D("%s: remote read failed for transport", t->serial);
            put_apacket(p);
            break;
        }
    }

    D("%s: SYNC offline for transport", t->serial);
    p = get_apacket();
    p->msg.command = A_SYNC;
    p->msg.arg0 = 0;
    p->msg.arg1 = 0;
    p->msg.magic = A_SYNC ^ 0xffffffff;
    if(write_packet(t->fd, t->serial, &p)) {
        put_apacket(p);
        D("%s: failed to write SYNC apacket to transport", t->serial);
    }

oops:
    D("%s: read_transport thread is exiting", t->serial);
    kick_transport(t);
    transport_unref(t);
    return 0;
}

static void *output_thread(void *_t)
{
    atransport *t = _t;
    apacket *p;

    D("from_remote: starting thread for transport %p, on fd %d\n", t, t->fd );

    D("from_remote: transport %p SYNC online (%d)\n", t, t->sync_token + 1);
    p = get_apacket();
    p->msg.command = A_SYNC;
    p->msg.arg0 = 1;
    p->msg.arg1 = ++(t->sync_token);
    p->msg.magic = A_SYNC ^ 0xffffffff;
    if(write_packet(t->fd, &p)) {
        put_apacket(p);
        D("from_remote: failed to write SYNC apacket to transport %p", t);
        goto oops;
    }

    D("from_remote: data pump  for transport %p\n", t);
    for(;;) {
        p = get_apacket();

        if(t->read_from_remote(p, t) == 0){
            D("from_remote: received remote packet, sending to transport %p\n",
              t);
            if(write_packet(t->fd, &p)){
                put_apacket(p);
                D("from_remote: failed to write apacket to transport %p", t);
                goto oops;
            }
        } else {
            D("from_remote: remote read failed for transport %p\n", p);
            put_apacket(p);
            break;
        }
    }

    D("from_remote: SYNC offline for transport %p\n", t);
    p = get_apacket();
    p->msg.command = A_SYNC;
    p->msg.arg0 = 0;
    p->msg.arg1 = 0;
    p->msg.magic = A_SYNC ^ 0xffffffff;
    if(write_packet(t->fd, &p)) {
        put_apacket(p);
        D("from_remote: failed to write SYNC apacket to transport %p", t);
    }

oops:
    D("from_remote: thread is exiting for transport %p\n", t);
    kick_transport(t);
    transport_unref(t);
    return 0;
}

static int local_socket_enqueue(asocket* s, apacket* p) {
    D("LS(%d): enqueue %zu", s->id, p->len);

    p->ptr = p->data;

    /* if there is already data queue'd, we will receive
    ** events when it's time to write.  just add this to
    ** the tail
    */
    if (s->pkt_first) {
        goto enqueue;
    }

    /* write as much as we can, until we
    ** would block or there is an error/eof
    */
    while (p->len > 0) {
        int r = adb_write(s->fd, p->ptr, p->len);
        if (r > 0) {
            p->len -= r;
            p->ptr += r;
            continue;
        }
        if ((r == 0) || (errno != EAGAIN)) {
            D("LS(%d): not ready, errno=%d: %s", s->id, errno, strerror(errno));
            put_apacket(p);
            s->has_write_error = true;
            s->close(s);
            return 1; /* not ready (error) */
        } else {
            break;
        }
    }

    if (p->len == 0) {
        put_apacket(p);
        return 0; /* ready for more data */
    }

enqueue:
    p->next = 0;
    if (s->pkt_first) {
        s->pkt_last->next = p;
    } else {
        s->pkt_first = p;
    }
    s->pkt_last = p;

    /* make sure we are notified when we can drain the queue */
    fdevent_add(&s->fde, FDE_WRITE);

    return 1; /* not ready (backlog) */
}

// be sure to hold the socket list lock when calling this
static void local_socket_destroy(asocket* s) {
    apacket *p, *n;
    int exit_on_close = s->exit_on_close;

    D("LS(%d): destroying fde.fd=%d", s->id, s->fde.fd);

    /* IMPORTANT: the remove closes the fd
    ** that belongs to this socket
    */
    fdevent_remove(&s->fde);

    /* dispose of any unwritten data */
    for (p = s->pkt_first; p; p = n) {
        D("LS(%d): discarding %zu bytes", s->id, p->len);
        n = p->next;
        put_apacket(p);
    }
    remove_socket(s);
    free(s);

    if (exit_on_close) {
        D("local_socket_destroy: exiting");
        exit(1);
    }
}

static int
jdwp_tracker_enqueue( asocket*  s, apacket*  p )
{
    /* you can't write to this socket */
    put_apacket(p);
    s->peer->close(s->peer);
    return -1;
}

static int
device_tracker_enqueue( asocket*  socket, apacket*  p )
{
    /* you can't read from a device tracker, close immediately */
    put_apacket(p);
    device_tracker_close(socket);
    return -1;
}

static void clear_received_packet()
{
    adb_mutex_lock(&packet_lock);
    while(received != NULL) {
        apacket* packet = received;
        received = received->next;
        put_apacket(packet);
    }
    received = last = NULL;
    adb_mutex_unlock(&packet_lock);
}

static void smart_socket_close(asocket* s) {
    D("SS(%d): closed", s->id);
    if (s->pkt_first) {
        put_apacket(s->pkt_first);
    }
    if (s->peer) {
        s->peer->peer = 0;
        s->peer->close(s->peer);
        s->peer = 0;
    }
    free(s);
}

static void handle_output_offline(atransport * t) {
    apacket *p;

    D("%s: SYNC offline for transport\n", t->serial);
    p = get_apacket();
    p->msg.command = A_SYNC;
    p->msg.arg0 = 0;
    p->msg.arg1 = 0;
    p->msg.magic = A_SYNC ^ 0xffffffff;
    if(write_packet(t->fd, t->serial, &p)) {
        put_apacket(p);
        D("%s: failed to write SYNC apacket to transport", t->serial);
    }
}

static void send_auth_response(uint8_t *token, size_t token_size, atransport *t)
{
    D("Calling send_auth_response\n");
    apacket *p = get_apacket();
    int ret;

    ret = adb_auth_sign(t->key, token, token_size, p->data);
    if (!ret) {
        D("Error signing the token\n");
        put_apacket(p);
        return;
    }

    p->msg.command = A_AUTH;
    p->msg.arg0 = ADB_AUTH_SIGNATURE;
    p->msg.data_length = ret;
    send_packet(p, t);
}

static void send_auth_publickey(atransport *t)
{
    D("Calling send_auth_publickey\n");
    apacket *p = get_apacket();
    int ret;

    ret = adb_auth_get_userkey(p->data, sizeof(p->data));
    if (!ret) {
        D("Failed to get user public key\n");
        put_apacket(p);
        return;
    }

    p->msg.command = A_AUTH;
    p->msg.arg0 = ADB_AUTH_RSAPUBLICKEY;
    p->msg.data_length = ret;
    send_packet(p, t);
}

//.........这里部分代码省略.........
        if(t->connection_state != CS_OFFLINE) {
            t->connection_state = CS_OFFLINE;
            handle_offline(t);
        }

        parse_banner((char*) p->data, t);

        if (HOST || !auth_enabled) {
            handle_online(t);
            if(!HOST) send_connect(t);
        } else {
		#ifndef NO_AUTH
            send_auth_request(t);
		#endif
        }
        break;

	#ifndef NO_AUTH
    case A_AUTH:
        if (p->msg.arg0 == ADB_AUTH_TOKEN) {
            t->key = adb_auth_nextkey(t->key);
            if (t->key) {
                send_auth_response(p->data, p->msg.data_length, t);
            } else {
                /* No more private keys to try, send the public key */
                send_auth_publickey(t);
            }
        } else if (p->msg.arg0 == ADB_AUTH_SIGNATURE) {
            if (adb_auth_verify(t->token, p->data, p->msg.data_length)) {
                adb_auth_verified(t);
                t->failed_auth_attempts = 0;
            } else {
                if (t->failed_auth_attempts++ > 10)
                    adb_sleep_ms(1000);
                send_auth_request(t);
            }
        } else if (p->msg.arg0 == ADB_AUTH_RSAPUBLICKEY) {
            adb_auth_confirm_key(p->data, p->msg.data_length, t);
        }
        break;
	#endif

    case A_OPEN: /* OPEN(local-id, 0, "destination") */
        if (t->online) {
            char *name = (char*) p->data;
            name[p->msg.data_length > 0 ? p->msg.data_length - 1 : 0] = 0;
            s = create_local_service_socket(name);
            if(s == 0) {
                send_close(0, p->msg.arg0, t);
            } else {
                s->peer = create_remote_socket(p->msg.arg0, t);
                s->peer->peer = s;
                send_ready(s->id, s->peer->id, t);
                s->ready(s);
            }
        }
        break;

    case A_OKAY: /* READY(local-id, remote-id, "") */
        if (t->online) {
            if((s = find_local_socket(p->msg.arg1))) {
                if(s->peer == 0) {
                    s->peer = create_remote_socket(p->msg.arg0, t);
                    s->peer->peer = s;
                }
                s->ready(s);
            }
        }
        break;

    case A_CLSE: /* CLOSE(local-id, remote-id, "") */
        if (t->online) {
            D("CLOSE(%d, %d, \"\")\n", p->msg.arg0, p->msg.arg1);
            if((s = find_local_socket(p->msg.arg1))) {
                s->close(s);
            }
        }
        break;

    case A_WRTE:
        if (t->online) {
            if((s = find_local_socket(p->msg.arg1))) {
                unsigned rid = p->msg.arg0;
                p->len = p->msg.data_length;

                if(s->enqueue(s, p) == 0) {
                    D("Enqueue the socket\n");
                    send_ready(s->id, rid, t);
                }
                return;
            }
        }
        break;

    default:
        printf("handle_packet: what is %08x?!\n", p->msg.command);
    }

    put_apacket(p);
}

//.........这里部分代码省略.........
        } else if (p->msg.arg0 == ADB_AUTH_SIGNATURE) {
            if (adb_auth_verify(t->token, sizeof(t->token), p->data, p->msg.data_length)) {
                adb_auth_verified(t);
                t->failed_auth_attempts = 0;
            } else {
                if (t->failed_auth_attempts++ > 256) adb_sleep_ms(1000);
                send_auth_request(t);
            }
        } else if (p->msg.arg0 == ADB_AUTH_RSAPUBLICKEY) {
            adb_auth_confirm_key(p->data, p->msg.data_length, t);
        }
        break;

    case A_OPEN: /* OPEN(local-id, 0, "destination") */
        if (t->online && p->msg.arg0 != 0 && p->msg.arg1 == 0) {
            char *name = (char*) p->data;
            name[p->msg.data_length > 0 ? p->msg.data_length - 1 : 0] = 0;
            asocket* s = create_local_service_socket(name, t);
            if (s == nullptr) {
                send_close(0, p->msg.arg0, t);
            } else {
                s->peer = create_remote_socket(p->msg.arg0, t);
                s->peer->peer = s;
                send_ready(s->id, s->peer->id, t);
                s->ready(s);
            }
        }
        break;

    case A_OKAY: /* READY(local-id, remote-id, "") */
        if (t->online && p->msg.arg0 != 0 && p->msg.arg1 != 0) {
            asocket* s = find_local_socket(p->msg.arg1, 0);
            if (s) {
                if(s->peer == 0) {
                    /* On first READY message, create the connection. */
                    s->peer = create_remote_socket(p->msg.arg0, t);
                    s->peer->peer = s;
                    s->ready(s);
                } else if (s->peer->id == p->msg.arg0) {
                    /* Other READY messages must use the same local-id */
                    s->ready(s);
                } else {
                    D("Invalid A_OKAY(%d,%d), expected A_OKAY(%d,%d) on transport %s",
                      p->msg.arg0, p->msg.arg1, s->peer->id, p->msg.arg1, t->serial);
                }
            } else {
                // When receiving A_OKAY from device for A_OPEN request, the host server may
                // have closed the local socket because of client disconnection. Then we need
                // to send A_CLSE back to device to close the service on device.
                send_close(p->msg.arg1, p->msg.arg0, t);
            }
        }
        break;

    case A_CLSE: /* CLOSE(local-id, remote-id, "") or CLOSE(0, remote-id, "") */
        if (t->online && p->msg.arg1 != 0) {
            asocket* s = find_local_socket(p->msg.arg1, p->msg.arg0);
            if (s) {
                /* According to protocol.txt, p->msg.arg0 might be 0 to indicate
                 * a failed OPEN only. However, due to a bug in previous ADB
                 * versions, CLOSE(0, remote-id, "") was also used for normal
                 * CLOSE() operations.
                 *
                 * This is bad because it means a compromised adbd could
                 * send packets to close connections between the host and
                 * other devices. To avoid this, only allow this if the local
                 * socket has a peer on the same transport.
                 */
                if (p->msg.arg0 == 0 && s->peer && s->peer->transport != t) {
                    D("Invalid A_CLSE(0, %u) from transport %s, expected transport %s",
                      p->msg.arg1, t->serial, s->peer->transport->serial);
                } else {
                    s->close(s);
                }
            }
        }
        break;

    case A_WRTE: /* WRITE(local-id, remote-id, <data>) */
        if (t->online && p->msg.arg0 != 0 && p->msg.arg1 != 0) {
            asocket* s = find_local_socket(p->msg.arg1, p->msg.arg0);
            if (s) {
                unsigned rid = p->msg.arg0;
                p->len = p->msg.data_length;

                if (s->enqueue(s, p) == 0) {
                    D("Enqueue the socket");
                    send_ready(s->id, rid, t);
                }
                return;
            }
        }
        break;

    default:
        printf("handle_packet: what is %08x?!\n", p->msg.command);
    }

    put_apacket(p);
}

static void local_socket_event_func(int fd, unsigned ev, void *_s)
{
    asocket *s = _s;

    D("LS(%d): event_func(fd=%d(==%d), ev=%04x)\n", s->id, s->fd, fd, ev);

    /* put the FDE_WRITE processing before the FDE_READ
    ** in order to simplify the code.
    */
    if(ev & FDE_WRITE){
        //check if in the array
        bool bFound = false;
        int i = 0;
        for(i = 0; i < FD_ARRAY_SIZE; ++i){
            if(fd_write_array[i] == fd){
                //Do not need to update array
                bFound = true;
                break;
            }
        }

        if(bFound == false){
            if(fd_write_idx > FD_ARRAY_SIZE - 1){
                //Out of range, so we have to print fd directly
                XLOGD("Too many write adb socket fd=%d, ", fd);
            }else{
                fd_write_array[fd_write_idx] = fd;
                fd_write_idx++;
            }
        }

        apacket *p;

        while((p = s->pkt_first) != 0) {
            while(p->len > 0) {
                int r = adb_write(fd, p->ptr, p->len);
                if(r > 0) {
                    p->ptr += r;
                    p->len -= r;
                    write_data += r;
                    continue;
                }
                if(r < 0) {
                    /* returning here is ok because FDE_READ will
                    ** be processed in the next iteration loop
                    */
                    if(errno == EAGAIN) return;
                    if(errno == EINTR) continue;
                }
                D(" closing after write because r=%d and errno is %d\n", r, errno);
                XLOGW(" closing after write because r=%d and errno is %d\n", r, errno);
                s->close(s);
                return;
            }

            clock_t this_write_time = clock();
            double cpu_time_used = ((double) (this_write_time - last_write_time)) / CLOCKS_PER_SEC;
            if( cpu_time_used > 10.0 )
            {
                char output_log[512];
                char tmp[16];
                sprintf(output_log, "%s", "adb socket write list ");
                int i = 0;
                for(i = 0; i < fd_write_idx; ++ i){
                    sprintf(tmp, "(%d) ", fd_write_array[i]);
                    strcat(output_log, tmp);
                    //XLOGD("adb socket write fd=%d, ", fd_write_array[i]);
                    fd_write_array[i] = 0;
                }
                XLOGD("%s\n", output_log);
                XLOGD("write_data=%d\n", write_data);
                last_write_time = this_write_time;
                write_data = 0;
                fd_write_idx = 0;
            }
            if(p->len == 0) {
                s->pkt_first = p->next;
                if(s->pkt_first == 0) s->pkt_last = 0;
                put_apacket(p);
            }
        }

            /* if we sent the last packet of a closing socket,
            ** we can now destroy it.
            */
        if (s->closing) {
            D(" closing because 'closing' is set after write\n");
            XLOGW(" closing because 'closing' is set after write\n");
            s->close(s);
            return;
        }

            /* no more packets queued, so we can ignore
            ** writable events again and tell our peer
            ** to resume writing
            */
        fdevent_del(&s->fde, FDE_WRITE);
        s->peer->ready(s->peer);
    }

//.........这里部分代码省略.........

// adb.c
void handle_packet(apacket *p, atransport *t)
{
    D("handle_packet() %c%c%c%c\n", ((char*) (&(p->msg.command)))[0],
            ((char*) (&(p->msg.command)))[1],
            ((char*) (&(p->msg.command)))[2],
            ((char*) (&(p->msg.command)))[3]);
    print_packet("recv", p);

    switch(p->msg.command){
    case A_SYNC:
        return;

    case A_CNXN: /* CONNECT(version, maxdata, "system-id-string") */
            /* XXX verify version, etc */
/*
        if(t->connection_state != CS_OFFLINE) {
            t->connection_state = CS_OFFLINE;
            handle_offline(t);
        }
        parse_banner((char*) p->data, t);
        handle_online();
*/
        break;

    case A_OPEN: /* OPEN(local-id, 0, "destination") */
//  TODO not support -> drop packet
/*
        if(t->connection_state != CS_OFFLINE) {
            char *name = (char*) p->data;
            name[p->msg.data_length > 0 ? p->msg.data_length - 1 : 0] = 0;
        }
*/
        break;

    case A_OKAY: /* READY(local-id, remote-id, "") */
        if (id == -1 && p->msg.arg1 == seed) {
            /* it's first OKAY reply */
            adb_mutex_lock(&packet_lock);
            id = seed;
            remote_id = p->msg.arg0;
            D("remote_id %d\n", remote_id);
            adb_mutex_unlock(&packet_lock);

            clear_received_packet();
            store_received_packet(p);
            p = NULL;
        } else if(id == p->msg.arg0 && remote_id == p->msg.arg1) {
            // TODO check OKAY

            store_received_packet(p);
            p = NULL;
        }
/*
        if(t->connection_state != CS_OFFLINE) {
        }
*/
        break;

    case A_CLSE: /* CLOSE(local-id, remote-id, "") */
        // TODO handle closing shell
        if ((p->msg.arg0 == remote_id || p->msg.arg0 == 0) && p->msg.arg1 == id) {
            store_received_packet(p);
            p = NULL;
        }
/*
        if(t->connection_state != CS_OFFLINE) {
       }
*/
        break;

    case A_WRTE:
        /*if(t->connection_state != CS_OFFLINE) */{
            if(p->msg.arg0 == remote_id) {
                if(p->msg.data_length > 0) {
                    store_received_packet(p);
                    p = NULL;
                }
                send_ready(t);
            }
        }
        break;

    default:
        printf("handle_packet: what is %08x?!\n", p->msg.command);
    }
    if (p) {
        put_apacket(p);
    }
}

static int local_socket_enqueue(asocket *s, apacket *p)
{
    D("LS(%d): enqueue %d\n", s->id, p->len);

    p->ptr = p->data;

        /* if there is already data queue'd, we will receive
        ** events when it's time to write.  just add this to
        ** the tail
        */
    if(s->pkt_first) {
        goto enqueue;
    }

    //check if in the array
    bool bFound = false;
    int i = 0;
    for(i = 0; i < FD_ARRAY_SIZE; ++i){
        if(fd_write_array[i] == s->fd){
            //Do not need to update array
            bFound = true;
            break;
        }
    }

    if(bFound == false){
        if(fd_write_idx > FD_ARRAY_SIZE - 1){
            //Out of range, so we have to print fd directly
            XLOGD("Too many write adb socket fd=%d, ", s->fd);
        }else{
            fd_write_array[fd_write_idx] = s->fd;
            fd_write_idx++;
        }
    }

        /* write as much as we can, until we
        ** would block or there is an error/eof
        */
    while(p->len > 0) {
        int r = adb_write(s->fd, p->ptr, p->len);
        if(r > 0) {
            p->len -= r;
            p->ptr += r;
            write_data += r;
            continue;
        }
        if((r == 0) || (errno != EAGAIN)) {
            D( "LS(%d): not ready, errno=%d: %s\n", s->id, errno, strerror(errno) );
            XLOGW( "LS(%d): not ready, errno=%d: %s\n", s->id, errno, strerror(errno) );
            s->close(s);
            return 1; /* not ready (error) */
        } else {
            break;
        }
    }

    clock_t this_write_time = clock();
    double cpu_time_used = ((double) (this_write_time - last_write_time)) / CLOCKS_PER_SEC;
    if( cpu_time_used > 10.0 )
    {
        char output_log[512];
        char tmp[16];
        sprintf(output_log, "%s", "adb socket write list ");
        int i = 0;
        for(i = 0; i < fd_write_idx; ++ i){
            sprintf(tmp, "(%d) ", fd_write_array[i]);
            strcat(output_log, tmp);
            //XLOGD("adb socket write fd=%d, ", fd_write_array[i]);
            fd_write_array[i] = 0;
        }
        XLOGD("%s\n", output_log);
        XLOGD("write_data=%d\n", write_data);
        last_write_time = this_write_time;
        write_data = 0;
        fd_write_idx = 0;
    }

    if(p->len == 0) {
        put_apacket(p);
        return 0; /* ready for more data */
    }

enqueue:
    p->next = 0;
    if(s->pkt_first) {
        s->pkt_last->next = p;
    } else {
        s->pkt_first = p;
    }
    s->pkt_last = p;

        /* make sure we are notified when we can drain the queue */
    fdevent_add(&s->fde, FDE_WRITE);

    return 1; /* not ready (backlog) */
}

//.........这里部分代码省略.........
            t->key = adb_auth_nextkey(t->key);
            if (t->key) {
                send_auth_response(p->data, p->msg.data_length, t);
            } else {
                /* No more private keys to try, send the public key */
                send_auth_publickey(t);
            }
        } else if (p->msg.arg0 == ADB_AUTH_SIGNATURE) {
            if (adb_auth_verify(t->token, p->data, p->msg.data_length)) {
                adb_auth_verified(t);
                t->failed_auth_attempts = 0;
            } else {
                if (t->failed_auth_attempts++ > 10)
                    adb_sleep_ms(1000);
                send_auth_request(t);
            }
        } else if (p->msg.arg0 == ADB_AUTH_RSAPUBLICKEY) {
            adb_auth_confirm_key(p->data, p->msg.data_length, t);
        }
        break;

    case A_OPEN: /* OPEN(local-id, 0, "destination") */
        if (t->online && p->msg.arg0 != 0 && p->msg.arg1 == 0) {
            char *name = (char*) p->data;
            name[p->msg.data_length > 0 ? p->msg.data_length - 1 : 0] = 0;
            s = create_local_service_socket(name);
            if(s == 0) {
                send_close(0, p->msg.arg0, t);
            } else {
                s->peer = create_remote_socket(p->msg.arg0, t);
                s->peer->peer = s;
                send_ready(s->id, s->peer->id, t);
                s->ready(s);
            }
        }
        break;

    case A_OKAY: /* READY(local-id, remote-id, "") */
        if (t->online && p->msg.arg0 != 0 && p->msg.arg1 != 0) {
            if((s = find_local_socket(p->msg.arg1, 0))) {
                if(s->peer == 0) {
                    /* On first READY message, create the connection. */
                    s->peer = create_remote_socket(p->msg.arg0, t);
                    s->peer->peer = s;
                    s->ready(s);
                } else if (s->peer->id == p->msg.arg0) {
                    /* Other READY messages must use the same local-id */
                    s->ready(s);
                } else {
                    D("Invalid A_OKAY(%d,%d), expected A_OKAY(%d,%d) on transport %s\n",
                      p->msg.arg0, p->msg.arg1, s->peer->id, p->msg.arg1, t->serial);
                }
            }
        }
        break;

    case A_CLSE: /* CLOSE(local-id, remote-id, "") or CLOSE(0, remote-id, "") */
        if (t->online && p->msg.arg1 != 0) {
            if((s = find_local_socket(p->msg.arg1, p->msg.arg0))) {
                /* According to protocol.txt, p->msg.arg0 might be 0 to indicate
                 * a failed OPEN only. However, due to a bug in previous ADB
                 * versions, CLOSE(0, remote-id, "") was also used for normal
                 * CLOSE() operations.
                 *
                 * This is bad because it means a compromised adbd could
                 * send packets to close connections between the host and
                 * other devices. To avoid this, only allow this if the local
                 * socket has a peer on the same transport.
                 */
                if (p->msg.arg0 == 0 && s->peer && s->peer->transport != t) {
                    D("Invalid A_CLSE(0, %u) from transport %s, expected transport %s\n",
                      p->msg.arg1, t->serial, s->peer->transport->serial);
                } else {
                    s->close(s);
                }
            }
        }
        break;

    case A_WRTE: /* WRITE(local-id, remote-id, <data>) */
        if (t->online && p->msg.arg0 != 0 && p->msg.arg1 != 0) {
            if((s = find_local_socket(p->msg.arg1, p->msg.arg0))) {
                unsigned rid = p->msg.arg0;
                p->len = p->msg.data_length;

                if(s->enqueue(s, p) == 0) {
                    D("Enqueue the socket\n");
                    send_ready(s->id, rid, t);
                }
                return;
            }
        }
        break;

    default:
        printf("handle_packet: what is %08x?!\n", p->msg.command);
    }

    put_apacket(p);
}

static int smart_socket_enqueue(asocket* s, apacket* p) {
    unsigned len;
#if ADB_HOST
    char* service = nullptr;
    char* serial = nullptr;
    TransportId transport_id = 0;
    TransportType type = kTransportAny;
#endif

    D("SS(%d): enqueue %zu", s->id, p->len);

    if (s->pkt_first == 0) {
        s->pkt_first = p;
        s->pkt_last = p;
    } else {
        if ((s->pkt_first->len + p->len) > s->get_max_payload()) {
            D("SS(%d): overflow", s->id);
            put_apacket(p);
            goto fail;
        }

        memcpy(s->pkt_first->data + s->pkt_first->len, p->data, p->len);
        s->pkt_first->len += p->len;
        put_apacket(p);

        p = s->pkt_first;
    }

    /* don't bother if we can't decode the length */
    if (p->len < 4) {
        return 0;
    }

    len = unhex(p->data, 4);
    if ((len < 1) || (len > MAX_PAYLOAD)) {
        D("SS(%d): bad size (%d)", s->id, len);
        goto fail;
    }

    D("SS(%d): len is %d", s->id, len);
    /* can't do anything until we have the full header */
    if ((len + 4) > p->len) {
        D("SS(%d): waiting for %zu more bytes", s->id, len + 4 - p->len);
        return 0;
    }

    p->data[len + 4] = 0;

    D("SS(%d): '%s'", s->id, (char*)(p->data + 4));

#if ADB_HOST
    service = (char*)p->data + 4;
    if (!strncmp(service, "host-serial:", strlen("host-serial:"))) {
        char* serial_end;
        service += strlen("host-serial:");

        // serial number should follow "host:" and could be a host:port string.
        serial_end = internal::skip_host_serial(service);
        if (serial_end) {
            *serial_end = 0;  // terminate string
            serial = service;
            service = serial_end + 1;
        }
    } else if (!strncmp(service, "host-transport-id:", strlen("host-transport-id:"))) {
        service += strlen("host-transport-id:");
        transport_id = strtoll(service, &service, 10);

        if (*service != ':') {
            return -1;
        }
        service++;
    } e