// File: worker/src/Channel/ChannelRequest.cpp  
absl::flat_hash_map<std::string, ChannelRequest::MethodId> ChannelRequest::string2MethodId =  
{  
    // 关闭 Worker 。通常发生在 worker 进程被关闭时，由 Worker 的析构函数触发。  
    { "worker.close",                                ChannelRequest::MethodId::WORKER_CLOSE                                     },  
    // 获取 Worker 转储。返回值包含 worker 进程 Id 和本 Worker 下的 Router 的 routerIds 集合。  
    { "worker.dump",                                 ChannelRequest::MethodId::WORKER_DUMP                                      },  
    // 获取 Worker 资源使用率。结构见 http://docs.libuv.org/en/v1.x/misc.html#c.uv_rusage_t 或 mediasoup-client 库。  
    { "worker.getResourceUsage",                     ChannelRequest::MethodId::WORKER_GET_RESOURCE_USAGE                        },  
    // 更新设置。具体是更新日志等级(logLevel)和日志标签(logTags)。返回值无具体内容。  
    { "worker.updateSettings",                       ChannelRequest::MethodId::WORKER_UPDATE_SETTINGS                           },  
    // 创建 Router。请求参数的 routerId 有调用方提供。返回值无具体内容。  
    { "worker.createRouter",                         ChannelRequest::MethodId::WORKER_CREATE_ROUTER                             },  
    // 关闭 Router。该消息由 Worker 处理。返回值无具体内容。  
    { "router.close",                                ChannelRequest::MethodId::ROUTER_CLOSE                                     },  
    // 获取 Router 转储。返回值包含本 Router 的 id、transportIds、rtpObserverIds、mapProducerIdConsumerIds、mapConsumerIdProducerId、mapProducerIdObserverIds、mapDataProducerIdDataConsumerIds 和 mapDataConsumerIdDataProducerId。  
    { "router.dump",                                 ChannelRequest::MethodId::ROUTER_DUMP                                      },  
    // 创建 WebRtcTransport。请求参数主要主要有监听 IP 和 Port（旧版可能无法指定端口而只能使用允许范围内的随机端口）；是否启用 UDP 和(或) TCP；是否首选 UDP 和(或) TCP。返回值包含客户端创建 Transport 所需参数，暂略。  
    { "router.createWebRtcTransport",                ChannelRequest::MethodId::ROUTER_CREATE_WEBRTC_TRANSPORT                   },  
    // 创建 PlainTransport。请求参数类似有创建 WebRtcTransport，但无需指定使用 UDP 还是 TCP，因为只支持 UDP。额外的 comedia 表示是否需要预先连接 Transport；rtcpMux 表示是否 RTCP 复用 RTP 端口；enableSrtp 表示是否使用 Srtp，true 则需要提供 srtpCryptoSuite 参数。返回值包含客户端创建 Transport 所需参数，暂略。  
    { "router.createPlainTransport",                 ChannelRequest::MethodId::ROUTER_CREATE_PLAIN_TRANSPORT                    },  
    // 创建 PipeTransport。请求参数类似有创建 WebRtcTransport，但无需指定使用 UDP 还是 TCP，因为只支持 UDP。额外的 enableRtx 是否启用 Rtx 重传。返回值包含打通管道所需参数，暂略。  
    { "router.createPipeTransport",                  ChannelRequest::MethodId::ROUTER_CREATE_PIPE_TRANSPORT                     },  
    // 创建 DirectTransport。请求参数不能将 enableSctp 设置为 true，因为没必要。也无需提供监听 IP 和 Port 等信息。返回值包含直连需参数，暂略。  
    { "router.createDirectTransport",                ChannelRequest::MethodId::ROUTER_CREATE_DIRECT_TRANSPORT                   },  
    // 创建 ActiveSpeakerObserver。 请求参数必须提供 interval 。返回会议中发言人的音量。  
    { "router.createActiveSpeakerObserver",          ChannelRequest::MethodId::ROUTER_CREATE_ACTIVE_SPEAKER_OBSERVER            },  
    // 创建 AudioLevelObserver。请求参数必须提供 interval，maxEntries 和 threshold 。返回会议中唯一主导说话者。  
    { "router.createAudioLevelObserver",             ChannelRequest::MethodId::ROUTER_CREATE_AUDIO_LEVEL_OBSERVER               },  
    // 关闭 Transport。该消息由 Router 处理。Transport 的关闭会触发连锁反应，以确保其上的 Consumer、Producer，其他人消费本 Transport 上的 Producer 的 Consumer(有点绕) 得以关闭。返回值无具体内容。  
    { "transport.close",                             ChannelRequest::MethodId::TRANSPORT_CLOSE                                  },  
    // 获取 Transport 转储。返回值包含本 Transport 的 Id、是否是 DirectTransport 的 direct 等。更多返回参数详见：Transport.cpp 的 void Transport::FillJson(json& jsonObject) const 方法。  
    { "transport.dump",                              ChannelRequest::MethodId::TRANSPORT_DUMP                                   },  
    // 获取 Transport 的状态值。返回值见：Transport.cpp 的 void Transport::FillJsonStats(json& jsonArray) 方法。  
    { "transport.getStats",                          ChannelRequest::MethodId::TRANSPORT_GET_STATS                              },  
    // 连接 Transport。DirectTransport 不会发送该请求，创建 PipeTransport 时如果参数 comedia 为 true 也无需发送该请求。返回值暂略。  
    { "transport.connect",                           ChannelRequest::MethodId::TRANSPORT_CONNECT                                },  
    // 设置最大传入码率。用于拥塞控制。返回值无具体内容。  
    { "transport.setMaxIncomingBitrate",             ChannelRequest::MethodId::TRANSPORT_SET_MAX_INCOMING_BITRATE               },  
    // 设置最大传出码率。用于拥塞控制。返回值无具体内容。  
    { "transport.setMaxOutgoingBitrate",             ChannelRequest::MethodId::TRANSPORT_SET_MAX_OUTGOING_BITRATE               },  
    // 重启 ICE。 仅 WebRtcTransport 会发送该请求。  
    { "transport.restartIce",                        ChannelRequest::MethodId::TRANSPORT_RESTART_ICE                            },  
    // 创建 Producer。返回值暂略。  
    { "transport.produce",                           ChannelRequest::MethodId::TRANSPORT_PRODUCE                                },  
    // 创建 Consumer。返回值暂略。  
    { "transport.consume",                           ChannelRequest::MethodId::TRANSPORT_CONSUME                                },  
    // 创建 DataProducer。返回值暂略。  
    { "transport.produceData",                       ChannelRequest::MethodId::TRANSPORT_PRODUCE_DATA                           },  
    // 创建 DataConsumer。返回值暂略。  
    { "transport.consumeData",                       ChannelRequest::MethodId::TRANSPORT_CONSUME_DATA                           },  
    // 启用 Transport 的跟踪事件。可选事件包括：probation 和 bwe。返回值无具体内容。  
    { "transport.enableTraceEvent",                  ChannelRequest::MethodId::TRANSPORT_ENABLE_TRACE_EVENT                     },  
    // 关闭 Producer。该消息由 Transport 处理。Producer 的关闭会触发连锁反应，以确消费本 Producer 的 Consumer 和监听本 Producer 的 RtpObserver 得以关闭。返回值无具体内容。  
    { "producer.close",                              ChannelRequest::MethodId::PRODUCER_CLOSE                                   },  
    // 获取 Producer 转储。返回值暂略。  
    { "producer.dump",                               ChannelRequest::MethodId::PRODUCER_DUMP                                    },  
    // 获取 Producer 状态。返回值暂略。  
    { "producer.getStats",                           ChannelRequest::MethodId::PRODUCER_GET_STATS                               },  
    // 暂停 Producer 的生产。返回值无具体内容。  
    { "producer.pause",                              ChannelRequest::MethodId::PRODUCER_PAUSE                                   },  
    // 恢复 Producer 的生产。返回值无具体内容。  
    { "producer.resume" ,                            ChannelRequest::MethodId::PRODUCER_RESUME                                  },  
    // 启用 Producer 的跟踪事件。可选事件包括：rtp、keyframe、nack、pli 和 fir。返回值无具体内容。  
    { "producer.enableTraceEvent",                   ChannelRequest::MethodId::PRODUCER_ENABLE_TRACE_EVENT                      },  
    // 关闭 Consumer。该消息由 Transport 处理。返回值无具体内容。  
    { "consumer.close",                              ChannelRequest::MethodId::CONSUMER_CLOSE                                   },  
    // 获取 Consumer 转储。返回值暂略。  
    { "consumer.dump",                               ChannelRequest::MethodId::CONSUMER_DUMP                                    },  
    // 获取 Consumer 状态。返回值暂略。  
    { "consumer.getStats",                           ChannelRequest::MethodId::CONSUMER_GET_STATS                               },  
    // 暂停 Consumer 的消费。返回值无具体内容。  
    { "consumer.pause",                              ChannelRequest::MethodId::CONSUMER_PAUSE                                   },  
    // 暂停 Consumer 的消费。返回值无具体内容。  
    { "consumer.resume",                             ChannelRequest::MethodId::CONSUMER_RESUME                                  },  
    // 设置 Consumer 的首选 Layers。仅对 SimulcastConsumer 和 SvcConsumer 有效。返回值无具体内容。  
    { "consumer.setPreferredLayers",                 ChannelRequest::MethodId::CONSUMER_SET_PREFERRED_LAYERS                    },  
    // 设置 Consumer 的码率优先级。返回值无具体内容。  
    { "consumer.setPriority",                        ChannelRequest::MethodId::CONSUMER_SET_PRIORITY                            },  
    // Consumer 请求关键帧。返回值无具体内容。  
    { "consumer.requestKeyFrame",                    ChannelRequest::MethodId::CONSUMER_REQUEST_KEY_FRAME                       },  
    // 启用 Consumer 的跟踪事件。可选事件包括：rtp、keyframe、nack、pli 和 fir。返回值无具体内容。  
    { "consumer.enableTraceEvent",                   ChannelRequest::MethodId::CONSUMER_ENABLE_TRACE_EVENT                      },  
    // 如下 DataProducer 和 DataProducer 的请求类似于 Producer 和 Consumer   
    { "dataProducer.close",                          ChannelRequest::MethodId::DATA_PRODUCER_CLOSE                              },  
    { "dataProducer.dump",                           ChannelRequest::MethodId::DATA_PRODUCER_DUMP                               },  
    { "dataProducer.getStats",                       ChannelRequest::MethodId::DATA_PRODUCER_GET_STATS                          },  
    { "dataConsumer.close",                          ChannelRequest::MethodId::DATA_CONSUMER_CLOSE                              },  
    { "dataConsumer.dump",                           ChannelRequest::MethodId::DATA_CONSUMER_DUMP                               },  
    { "dataConsumer.getStats",                       ChannelRequest::MethodId::DATA_CONSUMER_GET_STATS                          },  
    // 获取尚未发出的 Sctp buffer 数量。  
    { "dataConsumer.getBufferedAmount",              ChannelRequest::MethodId::DATA_CONSUMER_GET_BUFFERED_AMOUNT                },  
    // 降低 Sctp buffer 阈值。  
    { "dataConsumer.setBufferedAmountLowThreshold",  ChannelRequest::MethodId::DATA_CONSUMER_SET_BUFFERED_AMOUNT_LOW_THRESHOLD  },  
    // 关闭 RtpObserver。该消息由 Router 处理。返回值无具体内容。  
    { "rtpObserver.close",                           ChannelRequest::MethodId::RTP_OBSERVER_CLOSE                               },  
    // 暂停 RtpObserver 的监听。该消息由 Router 处理。返回值无具体内容。  
    { "rtpObserver.pause",                           ChannelRequest::MethodId::RTP_OBSERVER_PAUSE                               },  
    // 恢复 RtpObserver 的监听。该消息由 Router 处理。返回值无具体内容。  
    { "rtpObserver.resume",                          ChannelRequest::MethodId::RTP_OBSERVER_RESUME                              },  
    // 向 RtpObserver 新增 Producer。返回值无具体内容。  
    { "rtpObserver.addProducer",                     ChannelRequest::MethodId::RTP_OBSERVER_ADD_PRODUCER                        },  
    // 从 RtpObserver 移除 Producer。返回值无具体内容。  
    { "rtpObserver.removeProducer",                  ChannelRequest::MethodId::RTP_OBSERVER_REMOVE_PRODUCER                     }  
};

// File: worker/src/PayloadChannel/PayloadChannelRequest.cpp  
absl::flat_hash_map<std::string, PayloadChannelRequest::MethodId> PayloadChannelRequest::string2MethodId =  
{  
    // DataConsumer 发送  
    { "dataConsumer.send", PayloadChannelRequest::MethodId::DATA_CONSUMER_SEND },  
};

// File: worker/src/main.cpp  
// Ensure we are called by our Node library.  
if (!std::getenv("MEDIASOUP_VERSION"))  
{  
    MS_ERROR_STD("you don't seem to be my real father!");  
    std::_Exit(EXIT_FAILURE);  
}

// File: worker/src/lib.hpp  
extern "C" int mediasoup_worker_run(  
  int argc,  
  char* argv[],  
  const char* version,  
  int consumerChannelFd,  
  int producerChannelFd,  
  int payloadConsumeChannelFd,  
  int payloadProduceChannelFd,  
  ChannelReadFn channelReadFn,  
  ChannelReadCtx channelReadCtx,  
  ChannelWriteFn channelWriteFn,  
  ChannelWriteCtx channelWriteCtx,  
  PayloadChannelReadFn payloadChannelReadFn,  
  PayloadChannelReadCtx payloadChannelReadCtx,  
  PayloadChannelWriteFn payloadChannelWriteFn,  
  PayloadChannelWriteCtx payloadChannelWriteCtx);

this._child = child_process_1.spawn(
       // command
       spawnBin, 
       // args
       spawnArgs, 
       // options
       {
           env: {
               MEDIASOUP_VERSION: '3.6.7'
           },
           detached: false,
           // fd 0 (stdin)   : Just ignore it.
           // fd 1 (stdout)  : Pipe it for 3rd libraries that log their own stuff.
           // fd 2 (stderr)  : Same as stdout.
           // fd 3 (channel) : Producer Channel fd.
           // fd 4 (channel) : Consumer Channel fd.
           // fd 5 (channel) : Producer PayloadChannel fd.
           // fd 6 (channel) : Consumer PayloadChannel fd.
           stdio: ['ignore', 'pipe', 'pipe', 'pipe', 'pipe', 'pipe', 'pipe'],
           windowsHide: true
       });
       this._pid = this._child.pid;
       this._channel = new Channel_1.Channel({
           producerSocket: this._child.stdio[3],
           consumerSocket: this._child.stdio[4],
           pid: this._pid
       });
       this._payloadChannel = new PayloadChannel_1.PayloadChannel({
           // NOTE: TypeScript does not like more than 5 fds.
           // @ts-ignore
           producerSocket: this._child.stdio[5],
           // @ts-ignore
           consumerSocket: this._child.stdio[6]
       });

this._consumerSocket.on('data', (buffer) => {
   if (!this._recvBuffer) {
       this._recvBuffer = buffer;
   }
   else {
       this._recvBuffer = Buffer.concat([this._recvBuffer, buffer], this._recvBuffer.length + buffer.length);
   }
   if (this._recvBuffer.length > NS_PAYLOAD_MAX_LEN) {
       logger.error('receiving buffer is full, discarding all data into it');
       // Reset the buffer and exit.
       this._recvBuffer = undefined;
       return;
   }
   while (true) // eslint-disable-line no-constant-condition
    {
       let nsPayload;
       try {
           nsPayload = netstring.nsPayload(this._recvBuffer);
       }
       catch (error) {
           logger.error('invalid netstring data received from the worker process: %s', String(error));
           // Reset the buffer and exit.
           this._recvBuffer = undefined;
           return;
       }
       // Incomplete netstring message.
       if (nsPayload === -1)
           return;
       try {
           // We can receive JSON messages (Channel messages) or log strings.
           switch (nsPayload[0]) {
               // 123 = '{' (a Channel JSON messsage).
               case 123:
                   this._processMessage(JSON.parse(nsPayload.toString('utf8')));
                   break;
               // 68 = 'D' (a debug log).
               case 68:
                   logger.debug(`[pid:${pid}] ${nsPayload.toString('utf8', 1)}`);
                   break;
               // 87 = 'W' (a warn log).
               case 87:
                   logger.warn(`[pid:${pid}] ${nsPayload.toString('utf8', 1)}`);
                   break;
               // 69 = 'E' (an error log).
               case 69:
                   logger.error(`[pid:${pid} ${nsPayload.toString('utf8', 1)}`);
                   break;
               // 88 = 'X' (a dump log).
               case 88:
                   // eslint-disable-next-line no-console
                   console.log(nsPayload.toString('utf8', 1));
                   break;
               default:
                   // eslint-disable-next-line no-console
                   console.warn(`worker[pid:${pid}] unexpected data: %s`, nsPayload.toString('utf8', 1));
           }
       }
       catch (error) {
           logger.error('received invalid message from the worker process: %s', String(error));
       }
       // Remove the read payload from the buffer.
       this._recvBuffer =
           this._recvBuffer.slice(netstring.nsLength(this._recvBuffer));
       if (!this._recvBuffer.length) {
           this._recvBuffer = undefined;
           return;
       }
   }
});

#define MS_CLASS "mediasoup-worker"
// #define MS_LOG_DEV_LEVEL 3

#include "common.hpp"
#include "DepLibSRTP.hpp"
#include "DepLibUV.hpp"
#include "DepLibWebRTC.hpp"
#include "DepOpenSSL.hpp"
#include "DepUsrSCTP.hpp"
#include "Logger.hpp"
#include "MediaSoupErrors.hpp"
#include "Settings.hpp"
#include "Utils.hpp"
#include "Worker.hpp"
#include "Channel/Notifier.hpp"
#include "Channel/UnixStreamSocket.hpp"
#include "PayloadChannel/Notifier.hpp"
#include "PayloadChannel/UnixStreamSocket.hpp"
#include "RTC/DtlsTransport.hpp"
#include "RTC/SrtpSession.hpp"
#include <uv.h>
#include <cerrno>
#include <csignal>  // sigaction()
#include <cstdlib>  // std::_Exit(), std::genenv()
#include <iostream> // std::cerr, std::endl
#include <map>
#include <string>

static constexpr int ConsumerChannelFd{ 3 };
static constexpr int ProducerChannelFd{ 4 };
static constexpr int PayloadConsumerChannelFd{ 5 };
static constexpr int PayloadProducerChannelFd{ 6 };

void IgnoreSignals();

int main(int argc, char* argv[])
{
  // Ensure we are called by our Node library.
  if (std::getenv("MEDIASOUP_VERSION") == nullptr)
  {
    MS_ERROR_STD("you don't seem to be my real father!");

    std::_Exit(EXIT_FAILURE);
  }

  std::string version = std::getenv("MEDIASOUP_VERSION");

  // Initialize libuv stuff (we need it for the Channel).
  DepLibUV::ClassInit();

  // Channel socket (it will be handled and deleted by the Worker).
  Channel::UnixStreamSocket* channel{ nullptr };

  // PayloadChannel socket (it will be handled and deleted by the Worker).
  PayloadChannel::UnixStreamSocket* payloadChannel{ nullptr };

  try
  {
    channel = new Channel::UnixStreamSocket(ConsumerChannelFd, ProducerChannelFd);
  }
  catch (const MediaSoupError& error)
  {
    MS_ERROR_STD("error creating the Channel: %s", error.what());

    std::_Exit(EXIT_FAILURE);
  }

  try
  {
    payloadChannel =
      new PayloadChannel::UnixStreamSocket(PayloadConsumerChannelFd, PayloadProducerChannelFd);
  }
  catch (const MediaSoupError& error)
  {
    MS_ERROR_STD("error creating the RTC Channel: %s", error.what());

    std::_Exit(EXIT_FAILURE);
  }

  // Initialize the Logger.
  Logger::ClassInit(channel);

  try
  {
    Settings::SetConfiguration(argc, argv);
  }
  catch (const MediaSoupTypeError& error)
  {
    MS_ERROR_STD("settings error: %s", error.what());

    // 42 is a custom exit code to notify "settings error" to the Node library.
    std::_Exit(42);
  }
  catch (const MediaSoupError& error)
  {
    MS_ERROR_STD("unexpected settings error: %s", error.what());

    std::_Exit(EXIT_FAILURE);
  }

  MS_DEBUG_TAG(info, "starting mediasoup-worker process [version:%s]", version.c_str());

#if defined(MS_LITTLE_ENDIAN)
  MS_DEBUG_TAG(info, "little-endian CPU detected");
#elif defined(MS_BIG_ENDIAN)
  MS_DEBUG_TAG(info, "big-endian CPU detected");
#else
  MS_WARN_TAG(info, "cannot determine whether little-endian or big-endian");
#endif

#if defined(INTPTR_MAX) && defined(INT32_MAX) && (INTPTR_MAX == INT32_MAX)
  MS_DEBUG_TAG(info, "32 bits architecture detected");
#elif defined(INTPTR_MAX) && defined(INT64_MAX) && (INTPTR_MAX == INT64_MAX)
  MS_DEBUG_TAG(info, "64 bits architecture detected");
#else
  MS_WARN_TAG(info, "cannot determine 32 or 64 bits architecture");
#endif

  Settings::PrintConfiguration();
  DepLibUV::PrintVersion();

  try
  {
    // Initialize static stuff.
    DepOpenSSL::ClassInit();
    DepLibSRTP::ClassInit();
    DepUsrSCTP::ClassInit();
    DepLibWebRTC::ClassInit();
    Utils::Crypto::ClassInit();
    RTC::DtlsTransport::ClassInit();
    RTC::SrtpSession::ClassInit();
    Channel::Notifier::ClassInit(channel);
    PayloadChannel::Notifier::ClassInit(payloadChannel);

    // Ignore some signals.
    IgnoreSignals();

    // Run the Worker.
    Worker worker(channel, payloadChannel);

    // Free static stuff.
    DepLibUV::ClassDestroy();
    DepLibSRTP::ClassDestroy();
    Utils::Crypto::ClassDestroy();
    DepLibWebRTC::ClassDestroy();
    RTC::DtlsTransport::ClassDestroy();
    DepUsrSCTP::ClassDestroy();

    // Wait a bit so peding messages to stdout/Channel arrive to the Node
    // process.
    uv_sleep(200);

    std::_Exit(EXIT_SUCCESS);
  }
  catch (const MediaSoupError& error)
  {
    MS_ERROR_STD("failure exit: %s", error.what());

    std::_Exit(EXIT_FAILURE);
  }
}

void IgnoreSignals()
{
#ifndef _WIN32
  MS_TRACE();

  int err;
  struct sigaction act; // NOLINT(cppcoreguidelines-pro-type-member-init)

  // clang-format off
  std::map<std::string, int> ignoredSignals =
  {
    { "PIPE", SIGPIPE },
    { "HUP",  SIGHUP  },
    { "ALRM", SIGALRM },
    { "USR1", SIGUSR1 },
    { "USR2", SIGUSR2 }
  };
  // clang-format on

  act.sa_handler = SIG_IGN; // NOLINT(cppcoreguidelines-pro-type-cstyle-cast)
  act.sa_flags   = 0;
  err            = sigfillset(&act.sa_mask);

  if (err != 0)
    MS_THROW_ERROR("sigfillset() failed: %s", std::strerror(errno));

  for (auto& kv : ignoredSignals)
  {
    auto& sigName = kv.first;
    int sigId     = kv.second;

    err = sigaction(sigId, &act, nullptr);

    if (err != 0)
      MS_THROW_ERROR("sigaction() failed for signal %s: %s", sigName.c_str(), std::strerror(errno));
  }
#endif
}

// Channel socket (it will be handled and deleted by the Worker).
Channel::UnixStreamSocket* channel{ nullptr };

// PayloadChannel socket (it will be handled and deleted by the Worker).
PayloadChannel::UnixStreamSocket* payloadChannel{ nullptr };

try
{
  channel = new Channel::UnixStreamSocket(ConsumerChannelFd, ProducerChannelFd);
}
catch (const MediaSoupError& error)
{
  MS_ERROR_STD("error creating the Channel: %s", error.what());
  std::_Exit(EXIT_FAILURE);
}
try
{
  payloadChannel =
    new PayloadChannel::UnixStreamSocket(PayloadConsumerChannelFd, PayloadProducerChannelFd);
}
catch (const MediaSoupError& error)
{
  MS_ERROR_STD("error creating the RTC Channel: %s", error.what());
  std::_Exit(EXIT_FAILURE);
}

UnixStreamSocket::UnixStreamSocket(int consumerFd, int producerFd)
  : consumerSocket(consumerFd, NsMessageMaxLen, this), producerSocket(producerFd, NsMessageMaxLen)
{
  MS_TRACE_STD();
}

ConsumerSocket::ConsumerSocket(int fd, size_t bufferSize, Listener* listener)
  : ::UnixStreamSocket(fd, bufferSize, ::UnixStreamSocket::Role::CONSUMER), listener(listener)
{
  MS_TRACE_STD();
}

UnixStreamSocket::UnixStreamSocket(int fd, size_t bufferSize, UnixStreamSocket::Role role)
  : bufferSize(bufferSize), role(role)
{
  MS_TRACE_STD();

  int err;

  this->uvHandle       = new uv_pipe_t;
  this->uvHandle->data = static_cast<void*>(this);

  err = uv_pipe_init(DepLibUV::GetLoop(), this->uvHandle, 0);

  if (err != 0)
  {
    delete this->uvHandle;
    this->uvHandle = nullptr;

    MS_THROW_ERROR_STD("uv_pipe_init() failed: %s", uv_strerror(err));
  }

  err = uv_pipe_open(this->uvHandle, fd);

  if (err != 0)
  {
    uv_close(reinterpret_cast<uv_handle_t*>(this->uvHandle), static_cast<uv_close_cb>(onClose));

    MS_THROW_ERROR_STD("uv_pipe_open() failed: %s", uv_strerror(err));
  }

  if (this->role == UnixStreamSocket::Role::CONSUMER)
  {
    // Start reading.
    err = uv_read_start(
      reinterpret_cast<uv_stream_t*>(this->uvHandle),
      static_cast<uv_alloc_cb>(onAlloc),
      static_cast<uv_read_cb>(onRead));

    if (err != 0)
    {
      uv_close(reinterpret_cast<uv_handle_t*>(this->uvHandle), static_cast<uv_close_cb>(onClose));

      MS_THROW_ERROR_STD("uv_read_start() failed: %s", uv_strerror(err));
    }
  }

  // NOTE: Don't allocate the buffer here. Instead wait for the first uv_alloc_cb().
}

inline static void onRead(uv_stream_t* handle, ssize_t nread, const uv_buf_t* buf)
{
  auto* socket = static_cast<UnixStreamSocket*>(handle->data);
  if (socket)
    socket->OnUvRead(nread, buf);
}

inline void UnixStreamSocket::OnUvRead(ssize_t nread, const uv_buf_t* /*buf*/)
{
  MS_TRACE_STD();

  if (nread == 0)
    return;

  // Data received.
  if (nread > 0)
  {
    // Update the buffer data length.
    this->bufferDataLen += static_cast<size_t>(nread);
    // Notify the subclass.
    UserOnUnixStreamRead();
  }
  // Peer disconnected.
  else if (nread == UV_EOF || nread == UV_ECONNRESET)
  {
    this->isClosedByPeer = true;
    // Close local side of the pipe.
    Close();
    // Notify the subclass.
    UserOnUnixStreamSocketClosed();
  }
  // Some error.
  else
  {
    MS_ERROR_STD("read error, closing the pipe: %s", uv_strerror(nread));
    this->hasError = true;
    // Close the socket.
    Close();
    // Notify the subclass.
    UserOnUnixStreamSocketClosed();
  }
}

void ConsumerSocket::UserOnUnixStreamRead()
{
  MS_TRACE_STD();

  // Be ready to parse more than a single message in a single chunk.
  while (true)
  {
    if (IsClosed())
      return;

    size_t readLen = this->bufferDataLen - this->msgStart;
    char* msgStart = nullptr;
    size_t msgLen;
    int nsRet = netstring_read(
      reinterpret_cast<char*>(this->buffer + this->msgStart), readLen, &msgStart, &msgLen);

    if (nsRet != 0)
    {
      switch (nsRet)
      {
        case NETSTRING_ERROR_TOO_SHORT:
        {
          // Check if the buffer is full.
          if (this->bufferDataLen == this->bufferSize)
          {
            // First case: the incomplete message does not begin at position 0 of
            // the buffer, so move the incomplete message to the position 0.
            if (this->msgStart != 0)
            {
              std::memmove(this->buffer, this->buffer + this->msgStart, readLen);
              this->msgStart      = 0;
              this->bufferDataLen = readLen;
            }
            // Second case: the incomplete message begins at position 0 of the buffer.
            // The message is too big, so discard it.
            else
            {
              MS_ERROR_STD(
                "no more space in the buffer for the unfinished message being parsed, "
                "discarding it");

              this->msgStart      = 0;
              this->bufferDataLen = 0;
            }
          }

          // Otherwise the buffer is not full, just wait.
          return;
        }

        case NETSTRING_ERROR_TOO_LONG:
        {
          MS_ERROR_STD("NETSTRING_ERROR_TOO_LONG");

          break;
        }

        case NETSTRING_ERROR_NO_COLON:
        {
          MS_ERROR_STD("NETSTRING_ERROR_NO_COLON");

          break;
        }

        case NETSTRING_ERROR_NO_COMMA:
        {
          MS_ERROR_STD("NETSTRING_ERROR_NO_COMMA");

          break;
        }

        case NETSTRING_ERROR_LEADING_ZERO:
        {
          MS_ERROR_STD("NETSTRING_ERROR_LEADING_ZERO");

          break;
        }

        case NETSTRING_ERROR_NO_LENGTH:
        {
          MS_ERROR_STD("NETSTRING_ERROR_NO_LENGTH");

          break;
        }
      }

      // Error, so reset and exit the parsing loop.
      this->msgStart      = 0;
      this->bufferDataLen = 0;

      return;
    }

    // If here it means that msgStart points to the beginning of a message
    // with msgLen bytes length, so recalculate readLen.
    readLen =
      reinterpret_cast<const uint8_t*>(msgStart) - (this->buffer + this->msgStart) + msgLen + 1;

    this->listener->OnConsumerSocketMessage(this, msgStart, msgLen);

    // If there is no more space available in the buffer and that is because
    // the latest parsed message filled it, then empty the full buffer.
    if ((this->msgStart + readLen) == this->bufferSize)
    {
      this->msgStart      = 0;
      this->bufferDataLen = 0;
    }
    // If there is still space in the buffer, set the beginning of the next
    // parsing to the next position after the parsed message.
    else
    {
      this->msgStart += readLen;
    }

    // If there is more data in the buffer after the parsed message
    // then parse again. Otherwise break here and wait for more data.
    if (this->bufferDataLen > this->msgStart)
    {
      continue;
    }

    break;
  }
}

void UnixStreamSocket::OnConsumerSocketMessage(
    ConsumerSocket* /*consumerSocket*/, char* msg, size_t msgLen)
  {
    MS_TRACE_STD();

    try
    {
      json jsonMessage = json::parse(msg, msg + msgLen);
      auto* request    = new Channel::Request(this, jsonMessage);
      // Notify the listener.
      try
      {
        this->listener->OnChannelRequest(this, request);
      }
      catch (const MediaSoupTypeError& error)
      {
        request->TypeError(error.what());
      }
      catch (const MediaSoupError& error)
      {
        request->Error(error.what());
      }
      // Delete the Request.
      delete request;
    }
    catch (const json::parse_error& error)
    {
      MS_ERROR_STD("JSON parsing error: %s", error.what());
    }
    catch (const MediaSoupError& error)
    {
      MS_ERROR_STD("discarding wrong Channel request");
    }
  }

Request::Request(Channel::UnixStreamSocket* channel, json& jsonRequest) : channel(channel)
  {
    MS_TRACE();

    auto jsonIdIt = jsonRequest.find("id");

    if (jsonIdIt == jsonRequest.end() || !Utils::Json::IsPositiveInteger(*jsonIdIt))
      MS_THROW_ERROR("missing id");

    this->id = jsonIdIt->get<uint32_t>();

    auto jsonMethodIt = jsonRequest.find("method");

    if (jsonMethodIt == jsonRequest.end() || !jsonMethodIt->is_string())
      MS_THROW_ERROR("missing method");

    this->method = jsonMethodIt->get<std::string>();

    auto methodIdIt = Request::string2MethodId.find(this->method);

    if (methodIdIt == Request::string2MethodId.end())
    {
      Error("unknown method");

      MS_THROW_ERROR("unknown method '%s'", this->method.c_str());
    }

    this->methodId = methodIdIt->second;

    auto jsonInternalIt = jsonRequest.find("internal");

    if (jsonInternalIt != jsonRequest.end() && jsonInternalIt->is_object())
      this->internal = *jsonInternalIt;
    else
      this->internal = json::object();

    auto jsonDataIt = jsonRequest.find("data");

    if (jsonDataIt != jsonRequest.end() && jsonDataIt->is_object())
      this->data = *jsonDataIt;
    else
      this->data = json::object();
  }

this->listener->OnChannelRequest(this, request);

inline void Worker::OnChannelRequest(Channel::UnixStreamSocket* /*channel*/, Channel::Request* request)
{
  MS_TRACE();

  MS_DEBUG_DEV(
    "Channel request received [method:%s, id:%" PRIu32 "]", request->method.c_str(), request->id);

  switch (request->methodId)
  {
    case Channel::Request::MethodId::WORKER_DUMP:
    {
      json data = json::object();

      FillJson(data);

      request->Accept(data);

      break;
    }

    case Channel::Request::MethodId::WORKER_GET_RESOURCE_USAGE:
    {
      json data = json::object();

      FillJsonResourceUsage(data);

      request->Accept(data);

      break;
    }

    case Channel::Request::MethodId::WORKER_UPDATE_SETTINGS:
    {
      Settings::HandleRequest(request);

      break;
    }

    case Channel::Request::MethodId::WORKER_CREATE_ROUTER:
    {
      std::string routerId;

      // This may throw.
      SetNewRouterIdFromInternal(request->internal, routerId);

      auto* router = new RTC::Router(routerId);

      this->mapRouters[routerId] = router;

      MS_DEBUG_DEV("Router created [routerId:%s]", routerId.c_str());

      request->Accept();

      break;
    }

    case Channel::Request::MethodId::ROUTER_CLOSE:
    {
      // This may throw.
      RTC::Router* router = GetRouterFromInternal(request->internal);

      // Remove it from the map and delete it.
      this->mapRouters.erase(router->id);
      delete router;

      MS_DEBUG_DEV("Router closed [id:%s]", router->id.c_str());

      request->Accept();

      break;
    }

    // Any other request must be delivered to the corresponding Router.
    default:
    {
      // This may throw.
      RTC::Router* router = GetRouterFromInternal(request->internal);

      router->HandleRequest(request);

      break;
    }
  }
}

case Channel::Request::MethodId::WORKER_CREATE_ROUTER:
{
  std::string routerId;
  // This may throw.
  SetNewRouterIdFromInternal(request->internal, routerId);
  auto* router = new RTC::Router(routerId);
  this->mapRouters[routerId] = router;
  MS_DEBUG_DEV("Router created [routerId:%s]", routerId.c_str());
  request->Accept();
  break;
}