#include <boost/python.hpp>
#include <boost/thread.hpp>
#include <iostream>
#include <sstream>
#include <queue>

class Keeper
{
boost::python::object main_module;
boost::python::object main_namespace;

boost::mutex peacekeeper;

public:
    boost::mutex python_keeper;

    Keeper()
    {
        boost::mutex::scoped_lock l(this->peacekeeper);

        Py_Initialize();

        this->main_module = boost::python::import("__main__");
        this->main_namespace = main_module.attr("__dict__");

        try
        {
            boost::python::exec_file(boost::python::str("test.py"), this->main_namespace);
        } catch (boost::python::error_already_set const &)
        {
            PyErr_Print();
            abort;
        }
    };

    template <typename T>
    T Get(std::string object)
    {
        boost::mutex::scoped_lock l(this->peacekeeper);

        try
        {
            return boost::python::extract<T>(object, this->main_namespace);
        } catch (boost::python::error_already_set const &)
        {
            PyErr_Print();
            abort();
        }
    };
};

template<>
boost::python::object Keeper::Get<boost::python::object>(std::string object)
{
    boost::mutex::scoped_lock l(this->peacekeeper);

    try
    {
        return this->main_namespace[object];
    } catch (boost::python::error_already_set const &)
    {
        PyErr_Print();
        abort();
    }
}

class Queue
{
std::queue<int> kept;
boost::mutex peacekeeper;
int size_;

public:
    Queue()
    {
        this->size_ = 0;
    };

    void give(int i)
    {
        boost::mutex::scoped_lock l(this->peacekeeper);
        this->kept.push(i);

        this->size_ += 1;
    };

    int take()
    {
        boost::mutex::scoped_lock l(this->peacekeeper);

        int tmp = this->kept.front();
        this->kept.pop();

        this->size_ -= 1;

        return tmp;
    }

    int size()
    {
        return this->size_;
    };
};

class Producer
{
Queue *myqueue;
Keeper *k;
int id;

public:
    Producer(Queue *myqueue, Keeper *k, int id)
    {
        this->myqueue = myqueue;
        this->k = k;
        this->id = id;
    };

    void run()
    {
        int i = 0;
        while (true)
        {
            this->myqueue->give(i);

            boost::python::object writer = this->k->Get<boost::python::object>("write");

            std::ostringstream os;
            os << ">>> PRODUCER " << id << " give " << i << std::endl;

            {
                boost::mutex::scoped_lock l(this->k->python_keeper);
                writer(boost::python::str(os.str()));
            }

            boost::xtime xt;
            boost::xtime_get(&xt, boost::TIME_UTC);
            xt.sec += 1;

            i += 1;

            boost::thread::sleep(xt);
        }
    }
};

class Consumer
{
Queue *myqueue;
Keeper *k;

public:
    Consumer(Queue *myqueue, Keeper *k)
    {
        this->myqueue = myqueue;
        this->k = k;
    };

    void run()
    {
        while (true)
        {
            if (this->myqueue->size() > 0)
            {
                int tmp = this->myqueue->take();

                boost::python::object writer = this->k->Get<boost::python::object>("write");

                std::ostringstream os;
                os << "<<< CONSUMER get " << tmp << std::endl;

                {
                    boost::mutex::scoped_lock l(this->k->python_keeper);
                    writer(boost::python::str(os.str()));
                }
            }

            boost::xtime xt;
            boost::xtime_get(&xt, boost::TIME_UTC);
            xt.nsec += 10000;

            boost::thread::sleep(xt);
        }
    };
};

struct ConsumerThread
{
    Consumer *c;

    void operator()()
    {
        c->run();
    };
};

struct ProducerThread
{
    Producer *p;

    void operator()()
    {
        p->run();
    };
};

int main()
{
    Keeper k;
    Queue q;

    boost::thread_group grp;

    ConsumerThread cons;
    cons.c = new Consumer(&q, &k);
    grp.create_thread(cons);

    for (int i = 0; i < 3; i++)
    {
        ProducerThread prod;
        prod.p = new Producer(&q, &k, i);
        grp.create_thread(prod);
    }

    grp.join_all();
}