static void
PipeWatchProc(
    ClientData instanceData,	/* The pipe state. */
    int mask)			/* Events of interest; an OR-ed combination of
				 * TCL_READABLE, TCL_WRITABLE and
				 * TCL_EXCEPTION. */
{
    PipeState *psPtr = (PipeState *) instanceData;
    int newmask;

    if (psPtr->inFile) {
	newmask = mask & (TCL_READABLE | TCL_EXCEPTION);
	if (newmask) {
	    Tcl_CreateFileHandler(GetFd(psPtr->inFile), mask,
		    (Tcl_FileProc *) Tcl_NotifyChannel,
		    (ClientData) psPtr->channel);
	} else {
	    Tcl_DeleteFileHandler(GetFd(psPtr->inFile));
	}
    }
    if (psPtr->outFile) {
	newmask = mask & (TCL_WRITABLE | TCL_EXCEPTION);
	if (newmask) {
	    Tcl_CreateFileHandler(GetFd(psPtr->outFile), mask,
		    (Tcl_FileProc *) Tcl_NotifyChannel,
		    (ClientData) psPtr->channel);
	} else {
	    Tcl_DeleteFileHandler(GetFd(psPtr->outFile));
	}
    }
}

void
UnixSocketConsumerIO::OnConnected()
{
  MOZ_ASSERT(MessageLoopForIO::current() == GetIOLoop());
  MOZ_ASSERT(GetConnectionStatus() == SOCKET_IS_CONNECTED);

  if (!SetSocketFlags(GetFd())) {
    NS_WARNING("Cannot set socket flags!");
    FireSocketError();
    return;
  }

  if (!mConnector->SetUp(GetFd())) {
    NS_WARNING("Could not set up socket!");
    FireSocketError();
    return;
  }

  nsRefPtr<nsRunnable> r =
    new SocketIOEventRunnable<UnixSocketConsumerIO>(
      this, SocketIOEventRunnable<UnixSocketConsumerIO>::CONNECT_SUCCESS);
  NS_DispatchToMainThread(r);

  AddWatchers(READ_WATCHER, true);
  if (HasPendingData()) {
    AddWatchers(WRITE_WATCHER, false);
  }
}

Tcl_Channel
TclpCreateCommandChannel(
    TclFile readFile,		/* If non-null, gives the file for reading. */
    TclFile writeFile,		/* If non-null, gives the file for writing. */
    TclFile errorFile,		/* If non-null, gives the file where errors
				 * can be read. */
    int numPids,		/* The number of pids in the pid array. */
    Tcl_Pid *pidPtr)		/* An array of process identifiers. Allocated
				 * by the caller, freed when the channel is
				 * closed or the processes are detached (in a
				 * background exec). */
{
    char channelName[16 + TCL_INTEGER_SPACE];
    int channelId;
    PipeState *statePtr = (PipeState *) ckalloc((unsigned) sizeof(PipeState));
    int mode;

    statePtr->inFile = readFile;
    statePtr->outFile = writeFile;
    statePtr->errorFile = errorFile;
    statePtr->numPids = numPids;
    statePtr->pidPtr = pidPtr;
    statePtr->isNonBlocking = 0;

    mode = 0;
    if (readFile) {
	mode |= TCL_READABLE;
    }
    if (writeFile) {
	mode |= TCL_WRITABLE;
    }

    /*
     * Use one of the fds associated with the channel as the channel id.
     */

    if (readFile) {
	channelId = GetFd(readFile);
    } else if (writeFile) {
	channelId = GetFd(writeFile);
    } else if (errorFile) {
	channelId = GetFd(errorFile);
    } else {
	channelId = 0;
    }

    /*
     * For backward compatibility with previous versions of Tcl, we use
     * "file%d" as the base name for pipes even though it would be more
     * natural to use "pipe%d".
     */

    sprintf(channelName, "file%d", channelId);
    statePtr->channel = Tcl_CreateChannel(&pipeChannelType, channelName,
	    (ClientData) statePtr, mode);
    return statePtr->channel;
}

	void Close()
	{
		/* Remove ident socket from engine, and close it, but dont detatch it
		 * from its parent user class, or attempt to delete its memory.
		 */
		if (GetFd() > -1)
		{
			ServerInstance->Logs->Log(MODNAME, LOG_DEBUG, "Close ident socket %d", GetFd());
			SocketEngine::Close(this);
		}
	}

void WorldSocket::_HandlePing(WorldPacket* recvPacket)
{
    uint32 ping;
    if(recvPacket->size() < 4)
    {
        sLog.outString("Socket closed due to incomplete ping packet.");
        Disconnect();
        return;
    }

    *recvPacket >> ping;
    *recvPacket >> _latency;

    if(mSession)
    {
        mSession->_latency = _latency;
        mSession->m_lastPing = (uint32)UNIXTIME;

        // reset the move time diff calculator, don't worry it will be re-calculated next movement packet.
        mSession->m_clientTimeDelay = 0;
    }

#ifdef USING_BIG_ENDIAN
    swap32(&ping);
#endif

    OutPacket(SMSG_PONG, 4, &ping);

#ifdef WIN32
    // Dynamically change nagle buffering status based on latency.
    //if(_latency >= 250)
    // I think 350 is better, in a MMO 350 latency isn't that big that we need to worry about reducing the number of packets being sent.
    if(_latency >= 350)
    {
        if(!m_nagleEanbled)
        {
            u_long arg = 0;
            setsockopt(GetFd(), 0x6, 0x1, (const char*)&arg, sizeof(arg));
            m_nagleEanbled = true;
        }
    }
    else
    {
        if(m_nagleEanbled)
        {
            u_long arg = 1;
            setsockopt(GetFd(), 0x6, 0x1, (const char*)&arg, sizeof(arg));
            m_nagleEanbled = false;
        }
    }
#endif
}

	void Close()
	{
		/* Remove ident socket from engine, and close it, but dont detatch it
		 * from its parent user class, or attempt to delete its memory.
		 */
		if (GetFd() > -1)
		{
			ServerInstance->Logs->Log("m_ident", LOG_DEBUG, "Close ident socket %d", GetFd());
			ServerInstance->SE->DelFd(this);
			ServerInstance->SE->Close(GetFd());
			this->SetFd(-1);
		}
	}

	IdentRequestSocket(LocalUser* u) : user(u)
	{
		age = ServerInstance->Time();

		SetFd(socket(user->server_sa.sa.sa_family, SOCK_STREAM, 0));

		if (GetFd() == -1)
			throw ModuleException("Could not create socket");

		done = false;

		irc::sockets::sockaddrs bindaddr;
		irc::sockets::sockaddrs connaddr;

		memcpy(&bindaddr, &user->server_sa, sizeof(bindaddr));
		memcpy(&connaddr, &user->client_sa, sizeof(connaddr));

		if (connaddr.sa.sa_family == AF_INET6)
		{
			bindaddr.in6.sin6_port = 0;
			connaddr.in6.sin6_port = htons(113);
		}
		else
		{
			bindaddr.in4.sin_port = 0;
			connaddr.in4.sin_port = htons(113);
		}

		/* Attempt to bind (ident requests must come from the ip the query is referring to */
		if (SocketEngine::Bind(GetFd(), bindaddr) < 0)
		{
			this->Close();
			throw ModuleException("failed to bind()");
		}

		SocketEngine::NonBlocking(GetFd());

		/* Attempt connection (nonblocking) */
		if (SocketEngine::Connect(this, &connaddr.sa, connaddr.sa_size()) == -1 && errno != EINPROGRESS)
		{
			this->Close();
			throw ModuleException("connect() failed");
		}

		/* Add fd to socket engine */
		if (!SocketEngine::AddFd(this, FD_WANT_NO_READ | FD_WANT_POLL_WRITE))
		{
			this->Close();
			throw ModuleException("out of fds");
		}
	}

void
ListenSocketIO::OnSocketCanAcceptWithoutBlocking()
{
  MOZ_ASSERT(MessageLoopForIO::current() == GetIOLoop());
  MOZ_ASSERT(GetConnectionStatus() == SOCKET_IS_LISTENING);
  MOZ_ASSERT(mCOSocketIO);

  RemoveWatchers(READ_WATCHER|WRITE_WATCHER);

  struct sockaddr_storage storage;
  socklen_t addressLength = sizeof(storage);

  int fd;
  nsresult rv = mConnector->AcceptStreamSocket(
    GetFd(),
    reinterpret_cast<struct sockaddr*>(&storage), &addressLength,
    fd);
  if (NS_FAILED(rv)) {
    FireSocketError();
    return;
  }

  mCOSocketIO->Accept(fd,
                      reinterpret_cast<struct sockaddr*>(&storage),
                      addressLength);
}

static int
PipeInputProc(
    ClientData instanceData,	/* Pipe state. */
    char *buf,			/* Where to store data read. */
    int toRead,			/* How much space is available in the
				 * buffer? */
    int *errorCodePtr)		/* Where to store error code. */
{
    PipeState *psPtr = (PipeState *) instanceData;
    int bytesRead;		/* How many bytes were actually read from the
				 * input device? */

    *errorCodePtr = 0;

    /*
     * Assume there is always enough input available. This will block
     * appropriately, and read will unblock as soon as a short read is
     * possible, if the channel is in blocking mode. If the channel is
     * nonblocking, the read will never block. Some OSes can throw an
     * interrupt error, for which we should immediately retry. [Bug #415131]
     */

    do {
	bytesRead = read(GetFd(psPtr->inFile), buf, (size_t) toRead);
    } while ((bytesRead < 0) && (errno == EINTR));

    if (bytesRead < 0) {
	*errorCodePtr = errno;
	return -1;
    } else {
	return bytesRead;
    }
}

static int
PipeOutputProc(
    ClientData instanceData,	/* Pipe state. */
    const char *buf,		/* The data buffer. */
    int toWrite,		/* How many bytes to write? */
    int *errorCodePtr)		/* Where to store error code. */
{
    PipeState *psPtr = (PipeState *) instanceData;
    int written;

    *errorCodePtr = 0;

    /*
     * Some OSes can throw an interrupt error, for which we should immediately
     * retry. [Bug #415131]
     */

    do {
	written = write(GetFd(psPtr->outFile), buf, (size_t) toWrite);
    } while ((written < 0) && (errno == EINTR));

    if (written < 0) {
	*errorCodePtr = errno;
	return -1;
    } else {
	return written;
    }
}

void CConnect::Handle_Close()
{  
	CMessage *Msg = CMessageAlloctor::AllocMSG(0);
	Msg->fd_ = GetFd();
	EventLoop_->PushMsg(Msg);
	Remove();
}

static int
SetupStdFile(
    TclFile file,		/* File to dup, or NULL. */
    int type)			/* One of TCL_STDIN, TCL_STDOUT, TCL_STDERR */
{
    Tcl_Channel channel;
    int fd;
    int targetFd = 0;		/* Initializations here needed only to */
    int direction = 0;		/* prevent warnings about using uninitialized
				 * variables. */

    switch (type) {
    case TCL_STDIN:
	targetFd = 0;
	direction = TCL_READABLE;
	break;
    case TCL_STDOUT:
	targetFd = 1;
	direction = TCL_WRITABLE;
	break;
    case TCL_STDERR:
	targetFd = 2;
	direction = TCL_WRITABLE;
	break;
    }

    if (!file) {
	channel = Tcl_GetStdChannel(type);
	if (channel) {
	    file = TclpMakeFile(channel, direction);
	}
    }
    if (file) {
	fd = GetFd(file);
	if (fd != targetFd) {
	    if (dup2(fd, targetFd) == -1) {
		return 0;
	    }

	    /*
	     * Must clear the close-on-exec flag for the target FD, since some
	     * systems (e.g. Ultrix) do not clear the CLOEXEC flag on the
	     * target FD.
	     */

	    fcntl(targetFd, F_SETFD, 0);
	} else {
	    /*
	     * Since we aren't dup'ing the file, we need to explicitly clear
	     * the close-on-exec flag.
	     */

	    fcntl(fd, F_SETFD, 0);
	}
    } else {
	close(targetFd);
    }
    return 1;
}

  void Proceed(BluetoothStatus aStatus) override
  {
    DispatchBluetoothSocketHALResult(
      GetResultHandler(), &BluetoothSocketResultHandler::Connect,
      GetFd(), GetBdAddress(), GetConnectionStatus(), aStatus);

    MessageLoopForIO::current()->PostTask(
      FROM_HERE, new DeleteTask<ConnectWatcher>(this));
  }

static int
PipeGetHandleProc(
    ClientData instanceData,	/* The pipe state. */
    int direction,		/* TCL_READABLE or TCL_WRITABLE */
    ClientData *handlePtr)	/* Where to store the handle. */
{
    PipeState *psPtr = (PipeState *) instanceData;

    if (direction == TCL_READABLE && psPtr->inFile) {
	*handlePtr = (ClientData) INT2PTR(GetFd(psPtr->inFile));
	return TCL_OK;
    }
    if (direction == TCL_WRITABLE && psPtr->outFile) {
	*handlePtr = (ClientData) INT2PTR(GetFd(psPtr->outFile));
	return TCL_OK;
    }
    return TCL_ERROR;
}

void
UnixSocketWatcher::OnFileCanReadWithoutBlocking(int aFd)
{
  MOZ_ASSERT(MessageLoopForIO::current() == GetIOLoop());
  MOZ_ASSERT(aFd == GetFd());

  if (mConnectionStatus == SOCKET_IS_CONNECTED) {
    OnSocketCanReceiveWithoutBlocking();
  } else if (mConnectionStatus == SOCKET_IS_LISTENING) {
    int fd = TEMP_FAILURE_RETRY(accept(GetFd(), NULL, NULL));
    if (fd < 0) {
      OnError("accept", errno);
    } else {
      OnAccepted(fd);
    }
  } else {
    NS_NOTREACHED("invalid connection state for reading");
  }
}

void
BluetoothDaemonConnectionIO::OnSocketCanSendWithoutBlocking()
{
  MOZ_ASSERT(MessageLoopForIO::current() == GetIOLoop());
  MOZ_ASSERT(GetConnectionStatus() == SOCKET_IS_CONNECTED);
  MOZ_ASSERT(!IsShutdownOnIOThread());

  if (NS_WARN_IF(NS_FAILED(SendPendingData(GetFd())))) {
    RemoveWatchers(WRITE_WATCHER);
  }
}

void
BluetoothDaemonConnectionIO::OnSocketCanReceiveWithoutBlocking()
{
  ssize_t res = ReceiveData(GetFd());
  if (res < 0) {
    /* I/O error */
    RemoveWatchers(READ_WATCHER|WRITE_WATCHER);
  } else if (!res) {
    /* EOF or peer shutdown */
    RemoveWatchers(READ_WATCHER);
  }
}

void WorldSocket::_HandlePing(WorldPacket* recvPacket)
{
	uint32 ping;
	
	*recvPacket >> ping;
	*recvPacket >> _latency;

	if(mSession)
	{
		mSession->_latency = _latency;
		mSession->m_lastPing = (uint32)UNIXTIME;

		// reset the move time diff calculator, don't worry it will be re-calculated next movement packet.
		mSession->m_clientTimeDelay = 0;
	}

	OutPacket(SMSG_PONG, 4, &ping);

#ifdef WIN32
	// Dynamically change nagle buffering status based on latency.
	if(_latency >= 250)
	{
		if(!m_nagleEanbled)
		{
			u_long arg = 0;
			setsockopt(GetFd(), 0x6, 0x1, (const char*)&arg, sizeof(arg));
			m_nagleEanbled = true;
		}
	}
	else
	{
		if(m_nagleEanbled)
		{
			u_long arg = 1;
			setsockopt(GetFd(), 0x6, 0x1, (const char*)&arg, sizeof(arg));
			m_nagleEanbled = false;
		}
	}
#endif
}

void Socket::PostEvent(int events, bool oneshot)
{
    int kq = sSocketMgr.GetKq();

    struct kevent ev;
    if(oneshot)
        EV_SET(&ev, m_fd, events, EV_ADD | EV_ONESHOT, 0, 0, NULL);
    else
        EV_SET(&ev, m_fd, events, EV_ADD, 0, 0, NULL);

    if(kevent(kq, &ev, 1, 0, 0, NULL) < 0)
		Log.Warning("kqueue", "Could not modify event for fd %u", GetFd());
}

void JackSocketServerChannel::ClientRemove(detail::JackChannelTransactionInterface* socket_aux, int refnum)
{
    JackClientSocket* socket = dynamic_cast<JackClientSocket*>(socket_aux);
    assert(socket);
    int fd = GetFd(socket);
    assert(fd >= 0);

    jack_log("JackSocketServerChannel::ClientRemove ref = %d fd = %d", refnum, fd);
    fSocketTable.erase(fd);
    socket->Close();
    delete socket;
    fRebuild = true;
}