// RequestConnection.cpp

#include <new>

#include <OS.h>

#include "Channel.h"
#include "Connection.h"
#include "RequestChannel.h"
#include "RequestConnection.h"
#include "RequestHandler.h"

// DownStreamThread
class RequestConnection::DownStreamThread {
public:
        DownStreamThread()
                : fThread(-1),
                  fConnection(NULL),
                  fChannel(NULL),
                  fHandler(NULL),
                  fTerminating(false)
        {
        }

        ~DownStreamThread()
        {
                Terminate();
                delete fChannel;
        }

        status_t Init(RequestConnection* connection, Channel* channel,
                RequestHandler* handler)
        {
                if (!connection || !channel || !handler)
                        return B_BAD_VALUE;
                fConnection = connection;
                fHandler = handler;
                fChannel = new(std::nothrow) RequestChannel(channel);
                if (!fChannel)
                        return B_NO_MEMORY;
                fThread = spawn_thread(_LoopEntry, "down stream thread",
                        B_NORMAL_PRIORITY, this);
                if (fThread < 0)
                        return fThread;
                return B_OK;
        }

        void Run()
        {
                resume_thread(fThread);
        }

        void Terminate()
        {
                fTerminating = true;
                if (fThread > 0 && find_thread(NULL) != fThread) {
                        int32 result;
                        wait_for_thread(fThread, &result);
                }
        }

        RequestChannel* GetRequestChannel()
        {
                return fChannel;
        }

private:
        static int32 _LoopEntry(void* data)
        {
                return ((DownStreamThread*)data)->_Loop();
        }

        int32 _Loop()
        {
                while (!fTerminating) {
                        Request* request;
                        status_t error = fChannel->ReceiveRequest(&request);
                        if (error == B_OK) {
                                error = fHandler->HandleRequest(request, fChannel);
                                delete request;
                        }
                        if (error != B_OK)
                                fTerminating = fConnection->DownStreamChannelError(this, error);
                }
                return 0;
        }

private:
        thread_id                       fThread;
        RequestConnection*      fConnection;
        RequestChannel*         fChannel;
        RequestHandler*         fHandler;
        volatile bool           fTerminating;
};


// RequestConnection

// constructor
RequestConnection::RequestConnection(Connection* connection,
        RequestHandler* requestHandler, bool ownsRequestHandler)
        : fConnection(connection),
          fRequestHandler(requestHandler),
          fOwnsRequestHandler(ownsRequestHandler),
          fThreads(NULL),
          fThreadCount(0),
          fTerminationCount(0)
{
}

// destructor
RequestConnection::~RequestConnection()
{
        Close();
        delete[] fThreads;
        delete fConnection;
        if (fOwnsRequestHandler)
                delete fRequestHandler;
}

// Init
status_t
RequestConnection::Init()
{
        // check parameters
        if (!fConnection || !fRequestHandler)
                return B_BAD_VALUE;
        if (fConnection->CountDownStreamChannels() < 1)
                return B_ERROR;
        // create a thread per down-stream channel
        fThreadCount = fConnection->CountDownStreamChannels();
        fThreads = new(std::nothrow) DownStreamThread[fThreadCount];
        if (!fThreads)
                return B_NO_MEMORY;
        // initialize the threads
        for (int32 i = 0; i < fThreadCount; i++) {
                status_t error = fThreads[i].Init(this,
                        fConnection->DownStreamChannelAt(i), fRequestHandler);
                if (error != B_OK)
                        return error;
        }
        // run the threads
        for (int32 i = 0; i < fThreadCount; i++)
                fThreads[i].Run();
        return B_OK;
}

// Close
void
RequestConnection::Close()
{
        atomic_add(&fTerminationCount, 1);
        if (fConnection)
                fConnection->Close();
        if (fThreads) {
                for (int32 i = 0; i < fThreadCount; i++)
                        fThreads[i].Terminate();
        }
}

// SendRequest
status_t
RequestConnection::SendRequest(Request* request, Request** reply)
{
        return _SendRequest(request, reply, NULL);
}

// SendRequest
status_t
RequestConnection::SendRequest(Request* request, RequestHandler* replyHandler)
{
        if (!replyHandler)
                return B_BAD_VALUE;
        return _SendRequest(request, NULL, replyHandler);
}

// DownStreamChannelError
bool
RequestConnection::DownStreamChannelError(DownStreamThread* thread,
        status_t error)
{
        if (atomic_add(&fTerminationCount, 1) == 0 && fRequestHandler) {
                ConnectionBrokenRequest request;
                request.error = error;
                fRequestHandler->HandleRequest(&request, thread->GetRequestChannel());
        }
        return true;
}

// _SendRequest
status_t
RequestConnection::_SendRequest(Request* request, Request** _reply,
        RequestHandler* replyHandler)
{
        // check parameters
        if (!request)
                return B_BAD_VALUE;
        // get a channel
        Channel* channel = NULL;
        status_t error = fConnection->GetUpStreamChannel(&channel);
        if (error != B_OK)
                return error;
        // send the request
        RequestChannel requestChannel(channel);
        error = requestChannel.SendRequest(request);
        // receive the reply
        Request* reply = NULL;
        if (error == B_OK && (_reply || replyHandler)) {
                error = requestChannel.ReceiveRequest(&reply);
                // handle the reply
                if (error == B_OK) {
                        if (replyHandler)
                                error = replyHandler->HandleRequest(reply, &requestChannel);
                        if (error == B_OK && _reply)
                                *_reply = reply;
                        else
                                delete reply;
                }
        }
        // cleanup
        if (fConnection->PutUpStreamChannel(channel) != B_OK) {
                // Ugh! A serious error. Probably insufficient memory.
                delete channel;
        }
        return error;
}