#include "socktest.h"
using namespace LCBTest;
using std::string;
using std::vector;
class SockReadTest : public SockTest
{
};
namespace
{

/**
 * Set a specific 'rdwant' value. Send data smaller than the want, and then
 * send some more data.
 */
TEST_F(SockReadTest, testWant)
{
    ESocket sock;
    loop->connect(&sock);
    string expected("Hello World!");
    SendFuture sf(expected);

    ReadBreakCondition cond(&sock, expected.size());

    sock.reqrd(expected.size() * 2);
    sock.schedule();
    sock.conn->setSend(&sf);
    loop->setBreakCondition(&cond);
    loop->start();
    sf.wait();
    ASSERT_TRUE(cond.didBreak());

    // If everything is right, we should have nothing inside the buffer
    ASSERT_TRUE(sock.getReceived().empty());

    SendFuture sf2(expected);
    sock.conn->setSend(&sf2);
    cond = ReadBreakCondition(&sock, expected.size() * 2);
    loop->setBreakCondition(&cond);
    loop->start();
    sf2.wait();
    expected += expected;
    ASSERT_FALSE(sock.getReceived().empty());
    ASSERT_EQ(expected, sock.getReceived());
}

// Ensure the 'rdwant' flag is reset when we invoke the callback
TEST_F(SockReadTest, testWantReset)
{
    ESocket sock;
    loop->connect(&sock);
    string expected("Hi!!!");
    sock.reqrd(expected.size());
    SendFuture sf(expected);
    ReadBreakCondition rbc(&sock, expected.size());
    loop->setBreakCondition(&rbc);
    ASSERT_EQ(expected.size(), sock.ctx->rdwant);
    sock.schedule();
    sock.conn->setSend(&sf);
    loop->start();
    ASSERT_EQ(0, sock.ctx->rdwant);
}

/**
 * We should get an error if the socket is closed before we have data
 */
TEST_F(SockReadTest, testBrokenRead)
{
    ESocket sock;
    loop->connect(&sock);
    CloseFuture cf(CloseFuture::BEFORE_IO);
    sock.conn->setClose(&cf);
    sock.reqrd(5000);
    sock.schedule();
    ErrorBreakCondition ebc(&sock);
    loop->setBreakCondition(&ebc);
    loop->start();
    cf.wait();
    ASSERT_TRUE(sock.lasterr == LCB_ERR_NETWORK || sock.lasterr == LCB_ERR_SOCKET_SHUTDOWN);
}

TEST_F(SockReadTest, testReadAhead)
{
    ESocket sock;
    loop->connect(&sock);
    string sendStr(200, '$');
    unsigned wantSize = sendStr.size() / 2;
    SendFuture sf(sendStr);
    ReadBreakCondition rbc(&sock, wantSize);

    sock.reqrd(wantSize);
    sock.conn->setSend(&sf);
    sock.schedule();

    loop->setBreakCondition(&rbc);
    loop->start();
    sf.wait();

    ASSERT_TRUE(sock.getReceived().size() >= wantSize);
    if (sock.getReceived().size() == wantSize) {
        fprintf(stderr, "!!! received exactly wantsize. Slow network?\n");
    }
}

/**
 * Test the behavior of an orderly close where all the required data is
 * consumed.
 */
TEST_F(SockReadTest, testOrderlyClose)
{
    ESocket sock;
    loop->connect(&sock);
    string expected(200, '$');

    SendFuture sf(expected);
    ReadBreakCondition rbc(&sock, expected.size());
    CloseFuture cf(CloseFuture::AFTER_IO);

    sock.conn->setSend(&sf);
    sock.conn->setClose(&cf);
    loop->setBreakCondition(&rbc);
    sock.reqrd(expected.size());
    sock.schedule();
    loop->start();

    cf.wait();
    sf.wait();

    ASSERT_EQ(expected, sock.getReceived());
}

class ChunkReadActions : public IOActions
{
  public:
    int numChunks;
    vector< char > buffer;
    void onRead(ESocket *s, size_t nr)
    {
        lcbio_CTXRDITER iter;
        LCBIO_CTX_ITERFOR(s->ctx, &iter, nr)
        {
            unsigned nbuf = lcbio_ctx_risize(&iter);
            void *buf = lcbio_ctx_ribuf(&iter);
            numChunks++;
            buffer.insert(buffer.end(), (char *)buf, (char *)buf + nbuf);
        }
    }

    ChunkReadActions() : IOActions()
    {
        numChunks = 0;
    }
};

class CRABreakCondition : public BreakCondition
{
  public:
    ChunkReadActions *cra;
    int expected;
    CRABreakCondition(ChunkReadActions *actions, int exp)
    {
        cra = actions;
        expected = exp;
    }

    bool shouldBreakImpl()
    {
        return cra->numChunks >= expected;
    }
};
/** Tests the iterator chunking mechanism */
TEST_F(SockReadTest, testChunkedIter)
{
    // Set the chunked allocator
    ESocket sock;
    loop->connect(&sock);
    rdb_challoc(&sock.ctx->ior, rdb_chunkalloc_new(1));

    string toSend(20, '+');
    SendFuture sf(toSend);
    sock.conn->setSend(&sf);
    ChunkReadActions cra;
    CRABreakCondition bc(&cra, 20);

    sock.setActions(&cra);
    sock.reqrd(toSend.size());
    sock.schedule();
    loop->setBreakCondition(&bc);
    loop->start();

    ASSERT_EQ(toSend.size(), cra.numChunks);
    ASSERT_EQ(toSend.size(), cra.buffer.size());
    ASSERT_EQ(toSend, string(cra.buffer.begin(), cra.buffer.end()));
    sf.wait();
}
} // namespace