/*
 * Copyright 2008, Ingo Weinhold, ingo_weinhold@gmx.de.
 * Distributed under the terms of the MIT License.
 */


#include "UnixStreamEndpoint.h"

#include <stdio.h>
#include <sys/stat.h>

#include <AutoDeleter.h>

#include <vfs.h>

#include "UnixAddressManager.h"
#include "UnixFifo.h"


#define UNIX_STREAM_ENDPOINT_DEBUG_LEVEL        0
#define UNIX_DEBUG_LEVEL                                        UNIX_STREAM_ENDPOINT_DEBUG_LEVEL
#include "UnixDebug.h"


// Note on locking order (outermost -> innermost):
// UnixStreamEndpoint: connecting -> listening -> child
// -> UnixFifo (never lock more than one at a time)
// -> UnixAddressManager


UnixStreamEndpoint::UnixStreamEndpoint(net_socket* socket, bool atomic)
        :
        UnixEndpoint(socket),
        fPeerEndpoint(NULL),
        fReceiveFifo(NULL),
        fState(unix_stream_endpoint_state::Closed),
        fAcceptSemaphore(-1),
        fIsChild(false),
        fWasConnected(false),
        fAtomic(atomic)
{
        TRACE("[%" B_PRId32 "] %p->UnixStreamEndpoint::UnixStreamEndpoint()\n",
                find_thread(NULL), this);
}


UnixStreamEndpoint::~UnixStreamEndpoint()
{
        TRACE("[%" B_PRId32 "] %p->UnixStreamEndpoint::~UnixStreamEndpoint()\n",
                find_thread(NULL), this);
}


status_t
UnixStreamEndpoint::Init()
{
        TRACE("[%" B_PRId32 "] %p->UnixStreamEndpoint::Init()\n", find_thread(NULL),
                this);

        RETURN_ERROR(B_OK);
}


void
UnixStreamEndpoint::Uninit()
{
        TRACE("[%" B_PRId32 "] %p->UnixStreamEndpoint::Uninit()\n", find_thread(NULL),
                this);

        // check whether we're closed
        UnixStreamEndpointLocker locker(this);
        bool closed = (fState == unix_stream_endpoint_state::Closed);
        locker.Unlock();

        if (!closed) {
                // That probably means, we're a child endpoint of a listener and
                // have been fully connected, but not yet accepted. Our Close()
                // hook isn't called in this case. Do it manually.
                Close();
        }

        ReleaseReference();
}


status_t
UnixStreamEndpoint::Open()
{
        TRACE("[%" B_PRId32 "] %p->UnixStreamEndpoint::Open()\n", find_thread(NULL),
                this);

        status_t error = ProtocolSocket::Open();
        if (error != B_OK)
                RETURN_ERROR(error);

        fState = unix_stream_endpoint_state::NotConnected;

        RETURN_ERROR(B_OK);
}


status_t
UnixStreamEndpoint::Close()
{
        TRACE("[%" B_PRId32 "] %p->UnixStreamEndpoint::Close()\n", find_thread(NULL),
                this);

        UnixStreamEndpointLocker locker(this);

        if (fState == unix_stream_endpoint_state::Connected) {
                BReference<UnixStreamEndpoint> peerReference;
                UnixStreamEndpointLocker peerLocker;
                if (_LockConnectedEndpoints(locker, peerLocker) == B_OK) {
                        // The peer may be destroyed when we release it,
                        // so grab a reference for the locker's sake.
                        peerReference.SetTo(peerLocker.Get(), false);

                        // We're still connected. Disconnect both endpoints!
                        fPeerEndpoint->_Disconnect();
                        _Disconnect();
                }
        }

        if (fState == unix_stream_endpoint_state::Listening)
                _StopListening();

        _Unbind();

        fState = unix_stream_endpoint_state::Closed;
        RETURN_ERROR(B_OK);
}


status_t
UnixStreamEndpoint::Free()
{
        TRACE("[%" B_PRId32 "] %p->UnixStreamEndpoint::Free()\n", find_thread(NULL),
                this);

        UnixStreamEndpointLocker locker(this);

        _UnsetReceiveFifo();

        RETURN_ERROR(B_OK);
}


status_t
UnixStreamEndpoint::Bind(const struct sockaddr* _address)
{
        if (_address->sa_family != AF_UNIX)
                RETURN_ERROR(EAFNOSUPPORT);

        TRACE("[%" B_PRId32 "] %p->UnixStreamEndpoint::Bind(\"%s\")\n",
                find_thread(NULL), this,
                ConstSocketAddress(&gAddressModule, _address).AsString().Data());

        const sockaddr_un* address = (const sockaddr_un*)_address;

        UnixStreamEndpointLocker endpointLocker(this);

        if (fState != unix_stream_endpoint_state::NotConnected || IsBound())
                RETURN_ERROR(B_BAD_VALUE);

        RETURN_ERROR(_Bind(address));
}


status_t
UnixStreamEndpoint::Unbind()
{
        TRACE("[%" B_PRId32 "] %p->UnixStreamEndpoint::Unbind()\n", find_thread(NULL),
                this);

        UnixStreamEndpointLocker endpointLocker(this);

        RETURN_ERROR(_Unbind());
}


status_t
UnixStreamEndpoint::Listen(int backlog)
{
        TRACE("[%" B_PRId32 "] %p->UnixStreamEndpoint::Listen(%d)\n", find_thread(NULL),
                this, backlog);

        UnixStreamEndpointLocker endpointLocker(this);

        if (!IsBound())
                RETURN_ERROR(EDESTADDRREQ);
        if (fState != unix_stream_endpoint_state::NotConnected
                && fState != unix_stream_endpoint_state::Listening)
                RETURN_ERROR(EINVAL);

        gSocketModule->set_max_backlog(socket, backlog);

        if (fState == unix_stream_endpoint_state::NotConnected) {
                fAcceptSemaphore = create_sem(0, "unix accept");
                if (fAcceptSemaphore < 0)
                        RETURN_ERROR(ENOBUFS);

                _UnsetReceiveFifo();

                fCredentials.pid = getpid();
                fCredentials.uid = geteuid();
                fCredentials.gid = getegid();

                fState = unix_stream_endpoint_state::Listening;
        }

        RETURN_ERROR(B_OK);
}


status_t
UnixStreamEndpoint::Connect(const struct sockaddr* _address)
{
        if (_address->sa_family != AF_UNIX)
                RETURN_ERROR(EAFNOSUPPORT);

        TRACE("[%" B_PRId32 "] %p->UnixStreamEndpoint::Connect(\"%s\")\n",
                find_thread(NULL), this,
                ConstSocketAddress(&gAddressModule, _address).AsString().Data());

        const sockaddr_un* address = (const sockaddr_un*)_address;

        UnixStreamEndpointLocker endpointLocker(this);

        if (fState == unix_stream_endpoint_state::Connected)
                RETURN_ERROR(EISCONN);

        if (fState != unix_stream_endpoint_state::NotConnected)
                RETURN_ERROR(B_BAD_VALUE);
// TODO: If listening, we could set the backlog to 0 and connect.

        // check the address first
        UnixAddress unixAddress;

        if (address->sun_path[0] == '\0') {
                // internal address space (or empty address)
                int32 internalID;
                if (UnixAddress::IsEmptyAddress(*address))
                        RETURN_ERROR(B_BAD_VALUE);

                internalID = UnixAddress::InternalID(*address);
                if (internalID < 0)
                        RETURN_ERROR(internalID);

                unixAddress.SetTo(internalID);
        } else {
                // FS address space
                size_t pathLen = strnlen(address->sun_path, sizeof(address->sun_path));
                if (pathLen == 0 || pathLen == sizeof(address->sun_path))
                        RETURN_ERROR(B_BAD_VALUE);

                struct stat st;
                status_t error = vfs_read_stat(-1, address->sun_path, true, &st,
                        !gStackModule->is_syscall());
                if (error != B_OK)
                        RETURN_ERROR(error);

                if (!S_ISSOCK(st.st_mode))
                        RETURN_ERROR(B_BAD_VALUE);

                unixAddress.SetTo(st.st_dev, st.st_ino, NULL);
        }

        // get the peer endpoint
        UnixAddressManagerLocker addressLocker(gAddressManager);
        UnixEndpoint* listeningUnixEndpoint = gAddressManager.Lookup(unixAddress);
        if (listeningUnixEndpoint == NULL)
                RETURN_ERROR(ECONNREFUSED);
        UnixStreamEndpoint* listeningEndpoint
                = dynamic_cast<UnixStreamEndpoint*>(listeningUnixEndpoint);
        if (listeningEndpoint == NULL)
                RETURN_ERROR(EPROTOTYPE);
        BReference<UnixStreamEndpoint> peerReference(listeningEndpoint);
        addressLocker.Unlock();

        UnixStreamEndpointLocker peerLocker(listeningEndpoint);

        if (!listeningEndpoint->IsBound()
                || listeningEndpoint->fState != unix_stream_endpoint_state::Listening
                || listeningEndpoint->fAddress != unixAddress) {
                RETURN_ERROR(ECONNREFUSED);
        }

        // Allocate FIFOs for us and the socket we're going to spawn. We do that
        // now, so that the mess we need to cleanup, if allocating them fails, is
        // harmless.
        UnixFifoType type = fAtomic ? UnixFifoType::Datagram : UnixFifoType::Stream;
        UnixFifo* fifo = new(nothrow) UnixFifo(socket->receive.buffer_size, type);
        UnixFifo* peerFifo = new(nothrow) UnixFifo(socket->send.buffer_size, type);
        ObjectDeleter<UnixFifo> fifoDeleter(fifo);
        ObjectDeleter<UnixFifo> peerFifoDeleter(peerFifo);

        status_t error;
        if ((error = fifo->Init()) != B_OK || (error = peerFifo->Init()) != B_OK)
                return error;

        // spawn new endpoint for accept()
        net_socket* newSocket;
        error = gSocketModule->spawn_pending_socket(listeningEndpoint->socket,
                &newSocket);
        if (error != B_OK)
                RETURN_ERROR(error);

        // init connected peer endpoint
        UnixStreamEndpoint* connectedEndpoint = (UnixStreamEndpoint*)newSocket->first_protocol;

        UnixStreamEndpointLocker connectedLocker(connectedEndpoint);

        connectedEndpoint->_Spawn(this, listeningEndpoint, peerFifo);

        // update our attributes
        _UnsetReceiveFifo();

        fPeerEndpoint = connectedEndpoint;
        PeerAddress().SetTo(&connectedEndpoint->socket->address);
        fPeerEndpoint->AcquireReference();
        fReceiveFifo = fifo;

        fCredentials.pid = getpid();
        fCredentials.uid = geteuid();
        fCredentials.gid = getegid();

        fifoDeleter.Detach();
        peerFifoDeleter.Detach();

        fState = unix_stream_endpoint_state::Connected;
        fWasConnected = true;

        gSocketModule->set_connected(Socket());

        release_sem(listeningEndpoint->fAcceptSemaphore);

        connectedLocker.Unlock();
        peerLocker.Unlock();
        endpointLocker.Unlock();

        RETURN_ERROR(B_OK);
}


status_t
UnixStreamEndpoint::Accept(net_socket** _acceptedSocket)
{
        TRACE("[%" B_PRId32 "] %p->UnixStreamEndpoint::Accept()\n", find_thread(NULL),
                this);

        bigtime_t timeout = absolute_timeout(socket->receive.timeout);
        if (gStackModule->is_restarted_syscall())
                timeout = gStackModule->restore_syscall_restart_timeout();
        else
                gStackModule->store_syscall_restart_timeout(timeout);

        UnixStreamEndpointLocker locker(this);

        status_t error;
        do {
                locker.Unlock();

                error = acquire_sem_etc(fAcceptSemaphore, 1,
                        B_ABSOLUTE_TIMEOUT | B_CAN_INTERRUPT, timeout);
                if (error < B_OK)
                        break;

                locker.Lock();
                error = gSocketModule->dequeue_connected(socket, _acceptedSocket);
        } while (error != B_OK);

        if (error == B_TIMED_OUT && timeout == 0) {
                // translate non-blocking timeouts to the correct error code
                error = B_WOULD_BLOCK;
        }

        RETURN_ERROR(error);
}


ssize_t
UnixStreamEndpoint::Send(const iovec* vecs, size_t vecCount,
        ancillary_data_container* ancillaryData,
        const struct sockaddr* address, socklen_t addressLength, int flags)
{
        TRACE("[%" B_PRId32 "] %p->UnixStreamEndpoint::Send(%p, %ld, %p)\n",
                find_thread(NULL), this, vecs, vecCount, ancillaryData);

        // TODO: handle MSG_EOR
        bool atomic = (flags & MSG_EOR) != 0;
        if ((atomic && !fAtomic) || (flags & ~(MSG_DONTWAIT | MSG_EOR)) != 0)
                return EOPNOTSUPP;

        bigtime_t timeout = 0;
        if ((flags & MSG_DONTWAIT) == 0) {
                timeout = absolute_timeout(socket->send.timeout);
                if (gStackModule->is_restarted_syscall())
                        timeout = gStackModule->restore_syscall_restart_timeout();
                else
                        gStackModule->store_syscall_restart_timeout(timeout);
        }

        UnixStreamEndpointLocker locker(this);

        BReference<UnixStreamEndpoint> peerReference;
        UnixStreamEndpointLocker peerLocker;

        status_t error = _LockConnectedEndpoints(locker, peerLocker);
        if (error != B_OK)
                RETURN_ERROR(error);

        UnixStreamEndpoint* peerEndpoint = fPeerEndpoint;
        peerReference.SetTo(peerEndpoint);

        // lock the peer's FIFO
        UnixFifo* peerFifo = peerEndpoint->fReceiveFifo;
        BReference<UnixFifo> _(peerFifo);
        UnixFifoLocker fifoLocker(peerFifo);

        // unlock endpoints
        locker.Unlock();
        peerLocker.Unlock();

        ssize_t result = peerFifo->Write(vecs, vecCount, ancillaryData, NULL, timeout);

        // Notify select()ing readers, if we successfully wrote anything.
        size_t readable = peerFifo->Readable();
        bool notifyRead = (error == B_OK && readable > 0
                && !peerFifo->IsReadShutdown());

        // Notify select()ing writers, if we failed to write anything and there's
        // still room to write.
        size_t writable = peerFifo->Writable();
        bool notifyWrite = (error != B_OK && writable > 0
                && !peerFifo->IsWriteShutdown());

        // re-lock our endpoint (unlock FIFO to respect locking order)
        fifoLocker.Unlock();
        locker.Lock();

        bool peerLocked = (fPeerEndpoint == peerEndpoint
                && _LockConnectedEndpoints(locker, peerLocker) == B_OK);

        // send notifications
        if (peerLocked && notifyRead)
                gSocketModule->notify(peerEndpoint->socket, B_SELECT_READ, readable);
        if (notifyWrite)
                gSocketModule->notify(socket, B_SELECT_WRITE, writable);

        switch (result) {
                case UNIX_FIFO_SHUTDOWN:
                        if (fPeerEndpoint == peerEndpoint
                                        && fState == unix_stream_endpoint_state::Connected) {
                                // Orderly write shutdown on our side.
                                // Note: Linux and Solaris also send a SIGPIPE, but according
                                // the send() specification that shouldn't be done.
                                result = EPIPE;
                        } else {
                                // The FD has been closed.
                                result = EBADF;
                        }
                        break;
                case EPIPE:
                        // The socket module will generate SIGPIPE for us, if necessary.
                        break;
                case B_TIMED_OUT:
                        // Translate non-blocking timeouts to the correct error code.
                        if (timeout == 0)
                                result = B_WOULD_BLOCK;
                        break;
        }

        RETURN_ERROR(result);
}


ssize_t
UnixStreamEndpoint::Receive(const iovec* vecs, size_t vecCount,
        ancillary_data_container** _ancillaryData, struct sockaddr* _address,
        socklen_t* _addressLength, int flags)
{
        TRACE("[%" B_PRId32 "] %p->UnixStreamEndpoint::Receive(%p, %ld)\n",
                find_thread(NULL), this, vecs, vecCount);

        // TODO: handle MSG_WAITALL
        if ((flags & ~(MSG_DONTWAIT | MSG_PEEK | MSG_WAITALL)) != 0)
                return EOPNOTSUPP;

        bigtime_t timeout = 0;
        if ((flags & MSG_DONTWAIT) == 0) {
                timeout = absolute_timeout(socket->receive.timeout);
                if (gStackModule->is_restarted_syscall())
                        timeout = gStackModule->restore_syscall_restart_timeout();
                else
                        gStackModule->store_syscall_restart_timeout(timeout);
        }

        UnixStreamEndpointLocker locker(this);

        // We can read as long as we have a FIFO. I.e. we are still connected, or
        // disconnected and not yet reconnected/listening/closed.
        if (fReceiveFifo == NULL)
                RETURN_ERROR(ENOTCONN);

        UnixStreamEndpoint* peerEndpoint = fPeerEndpoint;
        BReference<UnixStreamEndpoint> peerReference(peerEndpoint);

        // Copy the peer address upfront. This way, if we read something, we don't
        // get into a potential race with Close().
        if (_address != NULL) {
                socklen_t addrLen = min_c(*_addressLength, socket->peer.ss_len);
                memcpy(_address, &socket->peer, addrLen);
                *_addressLength = addrLen;
        }

        // lock our FIFO
        UnixFifo* fifo = fReceiveFifo;
        BReference<UnixFifo> _(fifo);
        UnixFifoLocker fifoLocker(fifo);

        // unlock endpoint
        locker.Unlock();

        ssize_t result = fifo->Read(vecs, vecCount, _ancillaryData, NULL, timeout,
                (flags & MSG_PEEK) != 0);

        // Notify select()ing writers, if we successfully read anything.
        size_t writable = fifo->Writable();
        bool notifyWrite = (result >= 0 && writable > 0
                && !fifo->IsWriteShutdown());

        // Notify select()ing readers, if we failed to read anything and there's
        // still something left to read.
        size_t readable = fifo->Readable();
        bool notifyRead = (result < 0 && readable > 0
                && !fifo->IsReadShutdown());

        // re-lock our endpoint (unlock FIFO to respect locking order)
        fifoLocker.Unlock();
        locker.Lock();

        UnixStreamEndpointLocker peerLocker;
        bool peerLocked = (peerEndpoint != NULL && fPeerEndpoint == peerEndpoint
                && _LockConnectedEndpoints(locker, peerLocker) == B_OK);

        // send notifications
        if (notifyRead)
                gSocketModule->notify(socket, B_SELECT_READ, readable);
        if (peerLocked && notifyWrite)
                gSocketModule->notify(peerEndpoint->socket, B_SELECT_WRITE, writable);

        switch (result) {
                case UNIX_FIFO_SHUTDOWN:
                        // Either our socket was closed or read shutdown.
                        if (fState == unix_stream_endpoint_state::Closed) {
                                // The FD has been closed.
                                result = EBADF;
                        } else {
                                // if (fReceiveFifo == fifo) {
                                //              Orderly shutdown or the peer closed the connection.
                                // } else {
                                //              Weird case: Peer closed connection and we are already
                                //              reconnected (or listening).
                                // }
                                result = 0;
                        }
                        break;
                case B_TIMED_OUT:
                        // translate non-blocking timeouts to the correct error code
                        if (timeout == 0)
                                result = B_WOULD_BLOCK;
                        break;
        }

        RETURN_ERROR(result);
}


ssize_t
UnixStreamEndpoint::Sendable()
{
        TRACE("[%" B_PRId32 "] %p->UnixStreamEndpoint::Sendable()\n", find_thread(NULL),
                this);

        UnixStreamEndpointLocker locker(this);
        UnixStreamEndpointLocker peerLocker;

        status_t error = _LockConnectedEndpoints(locker, peerLocker);
        if (error != B_OK)
                RETURN_ERROR(error);

        // lock the peer's FIFO
        UnixFifo* peerFifo = fPeerEndpoint->fReceiveFifo;
        UnixFifoLocker fifoLocker(peerFifo);

        RETURN_ERROR(peerFifo->Writable());
}


ssize_t
UnixStreamEndpoint::Receivable()
{
        TRACE("[%" B_PRId32 "] %p->UnixStreamEndpoint::Receivable()\n", find_thread(NULL),
                this);

        UnixStreamEndpointLocker locker(this);

        if (fState == unix_stream_endpoint_state::Listening)
                return gSocketModule->count_connected(socket);

        if (fState != unix_stream_endpoint_state::Connected)
                RETURN_ERROR(ENOTCONN);

        UnixFifoLocker fifoLocker(fReceiveFifo);
        ssize_t readable = fReceiveFifo->Readable();
        if (readable == 0 && (fReceiveFifo->IsWriteShutdown()
                        || fReceiveFifo->IsReadShutdown())) {
                RETURN_ERROR(ENOTCONN);
        }
        RETURN_ERROR(readable);
}


status_t
UnixStreamEndpoint::SetReceiveBufferSize(size_t size)
{
        TRACE("[%" B_PRId32 "] %p->UnixStreamEndpoint::SetReceiveBufferSize(%lu)\n",
                find_thread(NULL), this, size);

        UnixStreamEndpointLocker locker(this);

        if (fReceiveFifo == NULL)
                return B_OK;

        UnixFifoLocker fifoLocker(fReceiveFifo);
        return fReceiveFifo->SetBufferCapacity(size);
}


status_t
UnixStreamEndpoint::GetPeerCredentials(ucred* credentials)
{
        UnixStreamEndpointLocker locker(this);
        UnixStreamEndpointLocker peerLocker;

        status_t error = _LockConnectedEndpoints(locker, peerLocker);
        if (error != B_OK)
                RETURN_ERROR(error);

        *credentials = fPeerEndpoint->fCredentials;

        return B_OK;
}


status_t
UnixStreamEndpoint::Shutdown(int direction)
{
        TRACE("[%" B_PRId32 "] %p->UnixStreamEndpoint::Shutdown(%d)\n",
                find_thread(NULL), this, direction);

        uint32 shutdown;
        uint32 peerShutdown;

        // translate the direction into shutdown flags for our and the peer fifo
        switch (direction) {
                case SHUT_RD:
                        shutdown = UNIX_FIFO_SHUTDOWN_READ;
                        peerShutdown = 0;
                        break;
                case SHUT_WR:
                        shutdown = 0;
                        peerShutdown = UNIX_FIFO_SHUTDOWN_WRITE;
                        break;
                case SHUT_RDWR:
                        shutdown = UNIX_FIFO_SHUTDOWN_READ;
                        peerShutdown = UNIX_FIFO_SHUTDOWN_WRITE;
                        break;
                default:
                        RETURN_ERROR(B_BAD_VALUE);
        }

        // lock endpoints
        UnixStreamEndpointLocker locker(this);
        UnixStreamEndpointLocker peerLocker;

        status_t error = _LockConnectedEndpoints(locker, peerLocker);
        if (error != B_OK)
                RETURN_ERROR(error);

        // shutdown our FIFO
        fReceiveFifo->Lock();
        fReceiveFifo->Shutdown(shutdown);
        fReceiveFifo->Unlock();

        // shutdown peer FIFO
        fPeerEndpoint->fReceiveFifo->Lock();
        fPeerEndpoint->fReceiveFifo->Shutdown(peerShutdown);
        fPeerEndpoint->fReceiveFifo->Unlock();

        // send select notifications
        if (direction == SHUT_RD || direction == SHUT_RDWR) {
                gSocketModule->notify(socket, B_SELECT_READ, EPIPE);
                gSocketModule->notify(fPeerEndpoint->socket, B_SELECT_WRITE, EPIPE);
        }
        if (direction == SHUT_WR || direction == SHUT_RDWR) {
                gSocketModule->notify(socket, B_SELECT_WRITE, EPIPE);
                gSocketModule->notify(fPeerEndpoint->socket, B_SELECT_READ, EPIPE);
        }

        RETURN_ERROR(B_OK);
}


void
UnixStreamEndpoint::_Spawn(UnixStreamEndpoint* connectingEndpoint,
        UnixStreamEndpoint* listeningEndpoint, UnixFifo* fifo)
{
        ProtocolSocket::Open();

        fIsChild = true;
        fPeerEndpoint = connectingEndpoint;
        fPeerEndpoint->AcquireReference();

        fReceiveFifo = fifo;

        PeerAddress().SetTo(&connectingEndpoint->socket->address);

        fCredentials = listeningEndpoint->fCredentials;

        fState = unix_stream_endpoint_state::Connected;

        gSocketModule->set_connected(Socket());
}


void
UnixStreamEndpoint::_Disconnect()
{
        // Both endpoints must be locked.

        // Write shutdown the receive FIFO.
        fReceiveFifo->Lock();
        fReceiveFifo->Shutdown(UNIX_FIFO_SHUTDOWN_WRITE);
        fReceiveFifo->Unlock();

        // select() notification.
        gSocketModule->notify(socket, B_SELECT_READ, ECONNRESET);
        gSocketModule->notify(socket, B_SELECT_WRITE, ECONNRESET);

        // Unset the peer endpoint.
        fPeerEndpoint->ReleaseReference();
        fPeerEndpoint = NULL;

        // We're officially disconnected.
// TODO: Deal with non accept()ed connections correctly!
        fIsChild = false;
        fState = unix_stream_endpoint_state::NotConnected;
}


status_t
UnixStreamEndpoint::_LockConnectedEndpoints(UnixStreamEndpointLocker& locker,
        UnixStreamEndpointLocker& peerLocker)
{
        if (fState != unix_stream_endpoint_state::Connected)
                RETURN_ERROR(fWasConnected ? EPIPE : ENOTCONN);

        // We need to lock the peer, too. Get a reference -- we might need to
        // unlock ourselves to get the locking order right.
        BReference<UnixStreamEndpoint> peerReference(fPeerEndpoint);
        UnixStreamEndpoint* peerEndpoint = fPeerEndpoint;

        if (fIsChild) {
                // We're the child, but locking order is the other way around.
                locker.Unlock();
                peerLocker.SetTo(peerEndpoint, false);
                locker.Lock();

                // recheck our state, also whether the peer is still the same
                if (fState != unix_stream_endpoint_state::Connected || peerEndpoint != fPeerEndpoint) {
                        peerLocker.Unset();
                        RETURN_ERROR(ENOTCONN);
                }
        } else
                peerLocker.SetTo(peerEndpoint, false);

        RETURN_ERROR(B_OK);
}


status_t
UnixStreamEndpoint::_Unbind()
{
        if (fState == unix_stream_endpoint_state::Connected
                || fState == unix_stream_endpoint_state::Listening)
                RETURN_ERROR(B_BAD_VALUE);

        if (IsBound())
                RETURN_ERROR(UnixEndpoint::_Unbind());

        RETURN_ERROR(B_OK);
}


void
UnixStreamEndpoint::_UnsetReceiveFifo()
{
        if (fReceiveFifo) {
                fReceiveFifo->ReleaseReference();
                fReceiveFifo = NULL;
        }
}


void
UnixStreamEndpoint::_StopListening()
{
        if (fState == unix_stream_endpoint_state::Listening) {
                delete_sem(fAcceptSemaphore);
                fAcceptSemaphore = -1;
                fState = unix_stream_endpoint_state::NotConnected;
        }
}