libcore/include/sirikata/core/network/RecordSSTStream.hpp

Go to the documentation of this file.

// Copyright (c) 2011 Sirikata Authors. All rights reserved.
// Use of this source code is governed by a BSD-style license that can
// be found in the LICENSE file.

#ifndef _SIRIKATA_CORE_RECORD_SST_STREAM_HPP_
#define _SIRIKATA_CORE_RECORD_SST_STREAM_HPP_

#include <sirikata/core/util/Platform.hpp>
#include <sirikata/core/network/Frame.hpp>
#include <sirikata/core/util/Liveness.hpp>
#include <sirikata/core/util/Time.hpp>

namespace Sirikata {

template<typename StreamPtrType>
class RecordSSTStream : public Liveness, public Noncopyable {
public:
    typedef std::tr1::function<void(MemoryReference)> RecordCallback;

    RecordSSTStream() {}
    ~RecordSSTStream() {
        destroy();
    }

    void initialize(StreamPtrType stream, RecordCallback cb) {
        mStream = stream;
        mCB = cb;

        using std::tr1::placeholders::_1;
        using std::tr1::placeholders::_2;

        assert(mStream);
        mStream->registerReadCallback(
            std::tr1::bind(
                &RecordSSTStream::handleRead, this,
                _1, _2
            )
        );
    }

    void write(const MemoryReference& data) {
        outstanding.push(Network::Frame::write(data.begin(), data.size()));
        writeSomeData(livenessToken());
    }

    void destroy() {
        letDie();

        if (!mStream) return;
        mStream->registerReadCallback(0);
    }
private:

    RecordSSTStream(const RecordSSTStream&);
    
    void handleRead(uint8* data, int size) {
        partial_frame.append((const char*)data, size);
        while(true) {
            String parsed = Network::Frame::parse(partial_frame);
            if (parsed.empty()) return;
            mCB( MemoryReference(parsed) );
        }
    }

    void writeSomeData(Liveness::Token alive) {
        if (!alive) return;

        static Duration retry_rate = Duration::milliseconds((int64)1);

        writing = true;

        if (!mStream) {
            // We're still waiting on the stream. Initialization should trigger
            // writing if it gets a stream and there's data waiting.
            writing = false;
            return;
        }

        // Otherwise, keep sending until we run out or
        while(!outstanding.empty()) {
            std::string& framed_msg = outstanding.front();
            int bytes_written = mStream->write((const uint8*)framed_msg.data(), framed_msg.size());
            if (bytes_written < 0) {
                // FIXME
                break;
            }
            else if (bytes_written < (int)framed_msg.size()) {
                framed_msg = framed_msg.substr(bytes_written);
                break;
            }
            else {
                outstanding.pop();
            }
        }

        if (outstanding.empty())
            writing = false;
        else
            mStream->getContext()->mainStrand->post(
                retry_rate,
                std::tr1::bind(&RecordSSTStream::writeSomeData, this, alive),
                "RecordSSTStream::writeSomeData"
            );
    }

    StreamPtrType mStream;
    RecordCallback mCB;

    // Outstanding data to be sent. FIXME efficiency
    std::queue<std::string> outstanding;
    // If writing is currently in progress
    bool writing;

    // Backlog of data, i.e. incomplete frame
    String partial_frame;
}; // class RecordSSTStream

} // namespace Sirikata

#endif //_SIRIKATA_CORE_RECORD_SST_STREAM_HPP_