1:   2:   3:   4:   5:   6:   7:   8:   9:  10:  11:  12:  13:  14:  15:  16:  17:  18:  19:  20:  21:  22:  23:  24:  25:  26:  27:  28:  29:  30:  31:  32:  33:  34:  35:  36:  37:  38:  39:  40:  41:  42:  43:  44:  45:  46:  47:  48:  49:  50:  51:  52:  53:  54:  55:  56:  57:  58:  59:  60:  61:  62:  63:  64:  65:  66:  67:  68:  69:  70:  71:  72:  73:  74:  75:  76:  77:  78:  79:  80:  81:  82:  83:  84:  85:  86:  87:  88:  89:  90:  91:  92:  93:  94:  95:  96:  97:  98:  99: 100: 101: 102: 103: 104: 105: 106: 107: 108: 109: 110: 111: 112: 113: 114: 115: 116: 117: 118: 119: 120: 121: 122: 123: 124: 125: 126: 127: 128: 129: 130: 131: 132: 133: 134: 135: 136: 137: 138: 139: 140: 141: 142: 143: 144: 145: 146: 147: 148: 149: 150: 151: 152: 153: 154: 155: 156: 157: 158: 159: 160: 161: 162: 163: 

<?php

    namespace Donut\Adapters;

    use Donut\IAdapter;
    use Donut\IProducer;
    use Donut\Message;
    use Donut\Time;
    use Nette\Utils\Json;
    use Ramsey\Uuid\Uuid;


    class DibiSqliteAdapter implements IAdapter
    {
        /** @var \Dibi\Connection */
        private $connection;


        /**
         * @param  string
         */
        public function __construct($file)
        {
            $this->connection = new \Dibi\Connection(array(
                'driver' => 'sqlite3',
                'file' => $file,
            ));
        }


        /**
         * {@inheritdoc}
         */
        public function createMessage(Message $message)
        {
            $this->connection->query('INSERT INTO [message]', array(
                'id%bin' => $message->getId()->getBytes(),
                'queue%s' => $message->getQueue(),
                'data%s' => Json::encode($message->getData()),
                'date%s' => $message->getDate()->getValue(),
                'created%s' => $message->getCreated()->getValue(),
                'status%i' => $message->getStatus(),
                'processed' => NULL,
            ));
        }


        /**
         * {@inheritdoc}
         */
        public function fetchMessage()
        {
            $row = $this->connection->fetch('SELECT * FROM [message] WHERE [status] = %i AND [processed] IS NULL ORDER BY [date] LIMIT 1', Message::STATUS_NEW);

            if ($row === FALSE) {
                return NULL;
            }

            return new Message(
                Uuid::fromBytes($row->id),
                $row->queue,
                Json::decode($row->data, Json::FORCE_ARRAY),
                Time::create($row->date),
                Time::create($row->created),
                $row->status,
                $row->processed ? Time::create($row->processed) : NULL
            );
        }


        /**
         * {@inheritdoc}
         */
        public function markAsDone(Message $message, Time $processed)
        {
            $this->connection->query('DELETE FROM [message] WHERE [id] = %bin', $message->getId()->getBytes());
        }


        /**
         * {@inheritdoc}
         */
        public function markAsFailed(Message $message, Time $processed)
        {
            $this->connection->query(
                'UPDATE [message] SET [status] = %i, [processed] = %s WHERE [id] = %bin',
                Message::STATUS_FAILED,
                $processed->getValue(),
                $message->getId()->getBytes()
            );
        }


        /**
         * {@inheritdoc}
         */
        public function log($subject, $text = NULL, Message $message = NULL, Time $date)
        {
            $this->connection->query('INSERT INTO [log]', array(
                'id%bin' => Uuid::uuid4()->getBytes(),
                'subject%s' => $subject,
                'text%s' => $text,
                'message_id%bin' => $message ? $message->getId()->getBytes() : NULL,
                'date%s' => $date->getValue(),
            ));
        }


        /**
         * {@inheritdoc}
         */
        public function getProducerLastRun(IProducer $producer)
        {
            $row = $this->connection->fetch('SELECT * FROM [producer] WHERE [producer] = %s', $producer->getUniqueId());

            if ($row && $row->lastrun !== NULL) {
                return Time::create($row->lastrun);
            }

            return NULL;
        }


        /**
         * {@inheritdoc}
         */
        public function saveProducerLastRun(IProducer $producer, Time $lastrun)
        {
            try {
                $this->connection->query('INSERT INTO [producer]', array(
                    'producer%s' => $producer->getUniqueId(),
                    'lastrun%s' => $lastrun->getValue(),
                ));

            } catch (\Dibi\UniqueConstraintViolationException $e) {
                $this->connection->query('UPDATE [producer] SET [lastrun] = %s WHERE [producer] = %s', $lastrun->getValue(), $producer->getUniqueId());
            }
        }


        /**
         * {@inheritdoc}
         */
        public function existsItem($group, $itemId)
        {
            $row = $this->connection->fetch('SELECT * FROM [item] WHERE [group_name] = %s AND [item] = %s', $group, $itemId);
            return $row !== FALSE && $row !== NULL;
        }


        /**
         * {@inheritdoc}
         */
        public function saveItem($group, $itemId)
        {
            $this->connection->query('INSERT INTO [item]', array(
                'group_name%s' => $group,
                'item%s' => $itemId,
                'date%s' => date('Y-m-d H:i:s')
            ));
        }
    }