深入學習 Node.js Net

$ nc -U /tmp/echo.sock # -U — Use UNIX domain socket

// 建立socket伺服器
const net = require("net");
let clients = 0;

const server = net.createServer(client => {
clients++;
let clientId = clients;
console.log("Client connect:", clientId);

client.on("end", () => {
console.log("Client disconnected:", clientId);
});

client.write("Welcome client: " + clientId + " \r\n");
client.pipe(client); // 把客戶端的資料返回給客戶端
});

server.listen(8000, () => {
console.log("Server started on port 8000");
});

// 建立socket客戶端
const net = require("net");

const client = net.connect(8000);

client.on("data", data => {
console.log(data.toString());
});

client.on("end", () => {
console.log("Client disconnected");
});

// 建立socket伺服器
const net = require("net");
let clients = 0;

const server = net.createServer(client => {
// 參考net基本使用相關程式碼
});

server.listen(8000, () => {
console.log("Server started on port 8000");
});

function createServer(options, connectionListener) {
return new Server(options, connectionListener);
}

function Server(options, connectionListener) {
if (!(this instanceof Server)) // 確保以new形式呼叫建構函式
return new Server(options, connectionListener);

EventEmitter.call(this);

if (typeof options === 'function') {
connectionListener = options;
options = {};
this.on('connection', connectionListener);
} else if (options == null || typeof options === 'object') {
options = options || {};

if (typeof connectionListener === 'function') {
this.on('connection', connectionListener);
}
}

this._connections = 0;
this._handle = null;
}
util.inherits(Server, EventEmitter);

function onconnection(err, clientHandle) {
var handle = this;
var self = handle.owner;

// 判斷是否超過最大連線數
if (self.maxConnections && self._connections >= self.maxConnections) {
clientHandle.close();
return;
}

// util.inherits(Socket, stream.Duplex);
var socket = new Socket({
handle: clientHandle,
allowHalfOpen: self.allowHalfOpen,
pauseOnCreate: self.pauseOnConnect
});
socket.readable = socket.writable = true;


self._connections++;

self.emit('connection', socket);
}

Server.prototype._listen2 = setupListenHandle;

server.listen(8000, () => {
console.log("Server started on port 8000");
});

Server.prototype.listen = function(...args) {
var normalized = normalizeArgs(args);
var options = normalized[0];
var cb = normalized[1];

var hasCallback = (cb !== null);
if (hasCallback) {
this.once('listening', cb);
}

options = options._handle || options.handle || options;

var backlog; 
// options.port：8000
if (typeof options.port === 'number' || typeof options.port === 'string') {
// start TCP server listening on host:port
if (options.host) {
lookupAndListen(this, options.port | 0, options.host, backlog,
options.exclusive);
} else { // Undefined host, listens on unspecified address
// Default addressType 4 will be used to search for master server
listenInCluster(this, null, options.port | 0, 4,
backlog, undefined, options.exclusive);
}
return this;
}
};

function listenInCluster(server, address, port, addressType,
backlog, fd, exclusive) {
// 如果exclusive是false（預設），則叢集的所有程序將使用相同的底層控制代碼，允許共享連線處理任務。
// 如果exclusive是true，則控制代碼不會被共享，如果嘗試埠共享將導致錯誤。
exclusive = !!exclusive;

// 引入cluster(叢集)模組
// Node.js在單個執行緒中執行單個例項。 使用者(開發者)為了使用現在的多核系統，有時候,
// 使用者(開發者)會用一串Node.js程序去處理負載任務。
if (cluster === null) cluster = require('cluster');

// 當該程序是主程序時，返回true。
if (cluster.isMaster || exclusive) {
// Will create a new handle
// _listen2 sets up the listened handle, it is still named like this
// to avoid breaking code that wraps this method
server._listen2(address, port, addressType, backlog, fd);
return;
}
}

// Server.prototype._listen2 = setupListenHandle; 
function setupListenHandle(address, port, addressType, backlog, fd) {
if (this._handle) {
debug('setupListenHandle: have a handle already');
} else {
debug('setupListenHandle: create a handle');

var rval = null;
if (rval === null)
rval = createServerHandle(address, port, addressType, fd);
this._handle = rval;
}

// 此處繫結onconnection函式
this._handle.onconnection = onconnection;
}

function createServerHandle(address, port, addressType, fd) {
var handle;

var isTCP = false;
// server.listen(handle[, backlog][, callback])
// 啟動一個伺服器，監聽已經繫結到埠、UNIX域套接字或Windows命名管道的給定控制代碼上的連線。
// 控制代碼物件可以是伺服器、套接字（任何具有底層_handle成員的東西），也可以是含有fd成員的物件，
// 該成員是一個有效的檔案描述符。
if (typeof fd === 'number' && fd >= 0) {
try {
handle = createHandle(fd, true);
} catch (e) {
// Not a fd we can listen on.This will trigger an error.
debug('listen invalid fd=%d:', fd, e.message);
return UV_EINVAL;
}
handle.open(fd);
handle.readable = true;
handle.writable = true;
} else if (port === -1 && addressType === -1) { // 處理UNIX domain socket或Windows pipe
handle = new Pipe(PipeConstants.SERVER);
} else {
handle = new TCP(TCPConstants.SERVER); // 建立TCP服務
isTCP = true;
}
return handle;
}

function createHandle(fd, is_server) {
// 基於檔案描述符確認handle的型別，TTY(文字終端)
const type = TTYWrap.guessHandleType(fd);
if (type === 'PIPE') {
return new Pipe(
is_server ? PipeConstants.SERVER : PipeConstants.SOCKET
);
}

if (type === 'TCP') {
return new TCP(
is_server ? TCPConstants.SERVER : TCPConstants.SOCKET
);
}
}

const { TCP, constants: TCPConstants } = process.binding('tcp_wrap');
const { Pipe, constants: PipeConstants } = process.binding('pipe_wrap');

const net = require("net");
const server = net.createServer(c => {
c.on("end", () => {
console.log("client disconnected");
});
c.write("hello\r\n");
c.pipe(c);
});
server.on("error", err => {
throw err;
});
// server.listen(path[, backlog][, callback]) for IPC servers
server.listen("/tmp/echo.sock", () => {
console.log("server bound");
});

$ nc -U /tmp/echo.sock

➜~ nc -U /tmp/echo.sock
hello
semlinker
semlinker
i love node
i love node