/**
 * Sends an e-mail using the selected settings
 *
 * @param {Object} mail Mail object
 * @param {Function} callback Callback function
 */
SMTPTransport.prototype.send = function (mail, callback) {
    this.getSocket(this.options, function (err, socketOptions) {
        if (err) {
            return callback(err);
        }

        var options = this.options;
        if (socketOptions && socketOptions.connection) {
            this.logger.info('Using proxied socket from %s:%s to %s:%s', socketOptions.connection.remoteAddress, socketOptions.connection.remotePort, options.host || '', options.port || '');
            // only copy options if we need to modify it
            options = assign(false, options);
            Object.keys(socketOptions).forEach(function (key) {
                options[key] = socketOptions[key];
            });
        }

        // 這里的 connection 沒有被釋放。
        var connection = new SMTPConnection(options);
        var returned = false;

        connection.once('error', function (err) {
            if (returned) {
                return;
            }
            returned = true;
            connection.close();
            return callback(err);
        });

        connection.once('end', function () {
            if (returned) {
                return;
            }
            returned = true;
            return callback(new Error('Connection closed'));
        });

        var sendMessage = function () {
            var envelope = mail.message.getEnvelope();
            var messageId = (mail.message.getHeader('message-id') || '').replace(/[<>\s]/g, '');
            var recipients = [].concat(envelope.to || []);
            if (recipients.length > 3) {
                recipients.push('...and ' + recipients.splice(2).length + ' more');
            }

            this.logger.info('Sending message <%s> to <%s>', messageId, recipients.join(', '));

            connection.send(envelope, mail.message.createReadStream(), function (err, info) {
                if (returned) {
                    return;
                }
                returned = true;

                connection.close();
                if (err) {
                    return callback(err);
                }
                info.envelope = {
                    from: envelope.from,
                    to: envelope.to
                };
                info.messageId = messageId;
                return callback(null, info);
            });
        }.bind(this);

        connection.connect(function () {
            if (returned) {
                return;
            }

            if (this.options.auth) {
                connection.login(this.options.auth, function (err) {
                    if (returned) {
                        return;
                    }

                    if (err) {
                        returned = true;
                        connection.close();
                        return callback(err);
                    }

                    sendMessage();
                });
            } else {
                sendMessage();
            }
        }.bind(this));
    }.bind(this));
};

Socket.prototype.end = function(data, encoding) {
 // 調(diào)用雙工流（可寫流）的 end 函數(shù)會觸發(fā) finish 事件。
  stream.Duplex.prototype.end.call(this, data, encoding);
  this.writable = false;
  // just in case we're waiting for an EOF.
  if (this.readable && !this._readableState.endEmitted)
    this.read(0);
  else
    maybeDestroy(this);
};

function maybeDestroy(socket) {
  if (!socket.readable &&
      !socket.writable &&
      !socket.destroyed &&
      !socket.connecting &&
      !socket._writableState.length) {
    // 這里調(diào)用的也是可寫流的 destroy 函數(shù)
    socket.destroy();
  }
}

// 可寫流 destroy 函數(shù)
function destroy(err, cb) {
   // 省略其余代碼
   // destroy 函數(shù)會調(diào)用 socket._destroy。
  this._destroy(err || null, (err) => {
    if (!cb && err) {
      process.nextTick(emitErrorNT, this, err);
      if (this._writableState) {
        this._writableState.errorEmitted = true;
      }
    } else if (cb) {
      cb(err);
    }
  });
}

Socket.prototype._destroy = function(exception, cb) {
  this.connecting = false;
  this.readable = this.writable = false;
  if (this._handle) {
    this[BYTES_READ] = this._handle.bytesRead;
    // this._handle = TCP()，調(diào)用TCP.close函數(shù)來關閉連接。
    this._handle.close(() => {
      debug('emit close');
      this.emit('close', isException);
    });
    this._handle.onread = noop;
    this._handle = null;
    this._sockname = null;
  }

  if (this._server) {
    COUNTER_NET_SERVER_CONNECTION_CLOSE(this);
    debug('has server');
    this._server._connections--;
    if (this._server._emitCloseIfDrained) {
      this._server._emitCloseIfDrained();
    }
  }
};

// socket 實現(xiàn)了Duplex，end 函數(shù)直接調(diào)用了 writableStream.end
Socket.prototype.end = function(data, encoding, callback) {
  stream.Duplex.prototype.end.call(this, data, encoding, callback);
  DTRACE_NET_STREAM_END(this);
  return this;
};

// _stream_writable.js
// writableStream.end 最終會調(diào)用如下函數(shù)
function finishMaybe(stream, state) {
  const need = needFinish(state);
  if (need) {
    prefinish(stream, state);
    if (state.pendingcb === 0) {
      state.finished = true;
      stream.emit('finish');

      // 這里的 state 存放可讀流的狀態(tài)變量
      // @node10 新增：autoDestroy 標志流是否在調(diào)用 end（）后自動調(diào)用自身的 destroy，在 v12 版本默認是 false。v14 版本開始默認為 true。
      // 所以當我們調(diào)用 socket.end（）的時候，不會立刻銷毀自己，僅僅會觸發(fā) finish 事件。
      if (state.autoDestroy) {
        const rState = stream._readableState;
        if (!rState || (rState.autoDestroy && rState.endEmitted)) {
          stream.destroy();
        }
      }
    }
  }
  return need;
}

// 那么 socket 什么時候會被銷毀呢？
// socket 構造函數(shù)
function Socket(options) {
     // 忽略
     // 注冊了end事件，觸發(fā)的時候這個函數(shù)會調(diào)用自己的 destroy。
     this.on('end', onReadableStreamEnd);
}

function onReadableStreamEnd() {
  // 省略
  if (!this.destroyed && !this.writable && !this.writableLength)
    // 同樣會調(diào)用可寫流的 destroy 然后調(diào)用 socket._destory（）
    this.destroy();
}

// Socket 的 end 事件是可讀流 read（）的時候觸發(fā)的。
// n 參數(shù)指定要讀取的特定字節(jié)數(shù)，如果不傳，每次返回內(nèi)部buffer中的全部數(shù)據(jù)。
Readable.prototype.read = function(n){
  const state = this._readableState;

  // 計算可以從緩沖區(qū)中讀取多少數(shù)據(jù)。
  n = howMuchToRead(n, state);

  // 本次可以讀取的字節(jié)數(shù)為0
  // 流內(nèi)部緩沖區(qū)buffer中的字節(jié)數(shù)為0
  // 可讀流的 ended 狀態(tài)為 true
  if (n === 0 && state.ended) {
    if (state.length === 0)
      // 結束自己
      endReadable(this);
    return null;
  }
}

function endReadable(stream) {
  const state = stream._readableState;
  debug('endReadable', state.endEmitted);
  if (!state.endEmitted) {
    state.ended = true;
    process.nextTick(endReadableNT, state, stream);
  }
}

function endReadableNT(state, stream) {
  debug('endReadableNT', state.endEmitted, state.length);
  if (!state.endEmitted && state.length === 0) {
    state.endEmitted = true;
    stream.readable = false;
    // 觸發(fā) stream（socket）的 end 事件。
    stream.emit('end');

    //這里和可寫流一樣也有個 autoDestroy 參數(shù)，同樣是默認 false。
    if (state.autoDestroy) {
      // In case of duplex streams we need a way to detect
      // if the writable side is ready for autoDestroy as well
      const wState = stream._writableState;
      if (!wState || (wState.autoDestroy && wState.finished)) {
        stream.destroy();
      }
    }
  }
}

SMTPConnection.prototype._onConnect = function () {
    // 省略
    // clear existing listeners for the socket
    this._socket.removeAllListeners('data');
    this._socket.removeAllListeners('timeout');
    this._socket.removeAllListeners('close');
    this._socket.removeAllListeners('end');
     // 省略
};