OutboxPersister::get() - Code Metrics - phpservicebus/persistence-doctrine2 - Measure and Improve Code Quality continuously with Scrutinizer

OutboxPersister::get() A
last analyzed 2020-01-22 18:29 UTC

↳ Parent: OutboxPersister

Complexity

Conditions	3
Paths	3

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Code Coverage

Tests	14
CRAP Score	3

Importance

Changes

Metric	Value
dl	0
loc	22
ccs	14
cts	14
cp	1
rs	9.568
c	0
b	0
f	0
cc	3
nc	3
nop	2
crap	3

<?php
namespace PSB\Persistence\Doctrine2\Outbox;


use Doctrine\DBAL\Connection;
use Doctrine\DBAL\ConnectionException;
use PSB\Core\Exception\CriticalErrorException;

class OutboxPersister
{
    /**
     * @var Connection
     */
    private $connection;

    /**
     * @var string
     */
    private $endpointsTableName;

    /**
     * @var string
     *
     */
    private $messagesTableName;

    /**
     * @param Connection $connection
     * @param string     $messagesTableName
     * @param string     $endpointsTableName
     */
    public function __construct(
        Connection $connection,
        $messagesTableName,
        $endpointsTableName
    ) {
        $this->connection = $connection;
        $this->endpointsTableName = $endpointsTableName;
        $this->messagesTableName = $messagesTableName;
    }

    /**
     * @param int    $endpointId
     * @param string $messageId
     *
     * @return array
     * @throws \Throwable
     */
    public function get($endpointId, $messageId)
    {
        try {
            $result = $this->connection->executeQuery(
                "SELECT * FROM {$this->messagesTableName} WHERE endpoint_id = ? AND message_id = ?",
                [$endpointId, hex2bin($this->stripDashes($messageId))]
            )->fetch(\PDO::FETCH_ASSOC);
        } catch (\Throwable $t) {
            throw $this->attemptToReconnectPresumedLostConnection($t);
        }

        if (!$result) {
            return null;
        }

        unset($result['id']);
        unset($result['dispatched_at']);
        unset($result['endpoint_id']);
        $result['message_id'] = bin2hex($result['message_id']);

        return $result;
    }

    /**
     * @param int   $endpointId
     * @param array $outboxRecord
     *
     * @throws \Exception
     */
    public function store($endpointId, array $outboxRecord)
    {
        $outboxRecord['message_id'] = hex2bin($this->stripDashes($outboxRecord['message_id']));
        $outboxRecord['endpoint_id'] = $endpointId;
        $outboxRecord['is_dispatched'] = 0;

        $this->connection->transactional(
            function (Connection $connection) use ($outboxRecord) {
                $connection->insert($this->messagesTableName, $outboxRecord);
            }
        );
    }

    /**
     * @param int    $endpointId
     * @param string $messageId
     *
     * @throws \Throwable
     */
    public function markAsDispatched($endpointId, $messageId)
    {
        try {
            $this->connection->executeUpdate(
                "UPDATE {$this->messagesTableName}
                     SET is_dispatched = 1, dispatched_at = ?, transport_operations = ''
                     WHERE endpoint_id = ? AND message_id = ? AND is_dispatched = 0",
                [
                    (new \DateTime('now', new \DateTimeZone('UTC')))->format('Y-m-d H:i:s'),
                    $endpointId,
                    hex2bin($this->stripDashes($messageId))
                ]
            );
        } catch (\Throwable $t) {
            throw $this->attemptToReconnectPresumedLostConnection($t);
        }
    }

    /**
     * Initiates the transaction
     * Attempts to reconnect once if disconnected.
     *
     * @return void
     * @throws \Throwable
     */
    public function beginTransaction()
    {
        try {
            $this->connection->beginTransaction();
        } catch (\Throwable $t) {
            throw $this->attemptToReconnectPresumedLostConnection($t);
        }
    }

    /**
     * Commits the transaction
     *
     * Does not attempt to reconnect if disconnected because the transaction would be broken anyway.
     * Reconnection should be done by rollBack.
     *
     * @return void
     * @throws ConnectionException
     */
    public function commit()
    {
        $this->connection->commit();
    }

    /**
     * Rolls back the transaction.
     * It makes sure that the connection is in the correct state regardless of what happened before.
     * Correct state means that connection is not rollback only and does not have a transaction nesting level > 0
     *
     * @throws \Throwable
     */
    public function rollBack()
    {
        try {
            /**
             * Roll back all the way as this is supposed to be the top level transaction and we want to reset
             * the nesting level
             */
            $transactionNestingLevel = $this->connection->getTransactionNestingLevel();
            for ($i = 0; $i < $transactionNestingLevel - 1; $i++) {
                $this->connection->rollBack();
            }
            $this->connection->rollBack();
        } catch (\Throwable $t) {
            $rethrowable = $this->attemptToReconnectPresumedLostConnection($t);
            /**
             * If connection is functional we need to make sure the connection is not rollback only.
             * This can only be achieved by starting a transaction and rolling it back (the "why" is found in
             * lines 1277-1279 of Doctrine\DBAL\Connection).
             */
            if ($rethrowable === $t) {
                $this->connection->beginTransaction();
                $this->connection->rollBack();
            }
            throw $rethrowable;
        }
    }

    /**
     * Attempts to reconnect once if disconnected.
     *
     * @param \DateTime $dateTime
     *
     * @throws \Exception
     */
    public function removeEntriesOlderThan(\DateTime $dateTime)
    {
        $dateTime->setTimezone(new \DateTimeZone('UTC'));
        $this->connection->executeUpdate(
            "DELETE FROM {$this->messagesTableName} WHERE is_dispatched = 1 AND dispatched_at <= ?",
            [$dateTime->format('Y-m-d H:i:s')]
        );
    }

    /**
     * @param string $endpointName
     *
     * @return int
     * @throws \Exception
     */
    public function fetchOrGenerateEndpointId($endpointName)
    {
        $endpointId = 0;
        $this->connection->transactional(
            function (Connection $connection) use ($endpointName, &$endpointId) {
                $lookupHash = md5($endpointName);
                $endpointRecord = $connection->executeQuery(
                    "SELECT * FROM {$this->endpointsTableName} WHERE lookup_hash = ?",
                    [hex2bin($lookupHash)]
                )->fetch(\PDO::FETCH_ASSOC);
                if (!$endpointRecord) {
                    $connection->insert(
                        $this->endpointsTableName,
                        ['lookup_hash' => hex2bin($lookupHash), 'name' => $endpointName]
                    );
                    $endpointId = (int)$connection->lastInsertId();
                } else {
                    $endpointId = (int)$endpointRecord['id'];
                }

            }
        );

        return $endpointId;
    }

    /**
     * @param string $messageId
     *
     * @return string
     */
    private function stripDashes($messageId)
    {
        return str_replace('-', '', $messageId);
    }

    /**
     * It attempts to reconnect if connection is non responsive. Failing to reconnect triggers a critical error.
     * If connection is responsive or successfully reconnected it rethrows, relying on the bus retries
     * to re-execute everything from the beginning.
     *
     * @param \Throwable $t
     *
     * @return \Throwable|CriticalErrorException
     */
    private function attemptToReconnectPresumedLostConnection(\Throwable $t)
    {
        // presumably, any exception caught here is related to some connection error
        if (!$this->connection->ping()) {
            // if pinging fails, we try to reconnect
            try {
                $this->connection->close();
                $this->connection->connect();
            } catch (\Throwable $t) {
                // if reconnecting fails, there is no way that the bus can continue to function
                return new CriticalErrorException("Database connection failed.", 0, $t);
            }
        }

        return $t;
    }
}


1		<?php
2		namespace PSB\Persistence\Doctrine2\Outbox;
3
4
5		use Doctrine\DBAL\Connection;
6		use Doctrine\DBAL\ConnectionException;
7		use PSB\Core\Exception\CriticalErrorException;
8
9		class OutboxPersister
10		{
11		/**
12		* @var Connection
13		*/
14		private $connection;
15
16		/**
17		* @var string
18		*/
19		private $endpointsTableName;
20
21		/**
22		* @var string
23		*
24		*/
25		private $messagesTableName;
26
27		/**
28		* @param Connection $connection
29		* @param string $messagesTableName
30		* @param string $endpointsTableName
31		*/
32	15	public function __construct(
33		Connection $connection,
34		$messagesTableName,
35		$endpointsTableName
36		) {
37	15	$this->connection = $connection;
38	15	$this->endpointsTableName = $endpointsTableName;
39	15	$this->messagesTableName = $messagesTableName;
40	15	}
41
42		/**
43		* @param int $endpointId
44		* @param string $messageId
45		*
46		* @return array
47		* @throws \Throwable
48		*/
49	7	public function get($endpointId, $messageId)
50		{
51		try {
52	7	$result = $this->connection->executeQuery(
53	7	"SELECT * FROM {$this->messagesTableName} WHERE endpoint_id = ? AND message_id = ?",
54	7	[$endpointId, hex2bin($this->stripDashes($messageId))]
55	6	)->fetch(\PDO::FETCH_ASSOC);
56	1	} catch (\Throwable $t) {
57	1	throw $this->attemptToReconnectPresumedLostConnection($t);
58		}
59
60	6	if (!$result) {
61	2	return null;
62		}
63
64	4	unset($result['id']);
65	4	unset($result['dispatched_at']);
66	4	unset($result['endpoint_id']);
67	4	$result['message_id'] = bin2hex($result['message_id']);
68
69	4	return $result;
70		}
71
72		/**
73		* @param int $endpointId
74		* @param array $outboxRecord
75		*
76		* @throws \Exception
77		*/
78	6	public function store($endpointId, array $outboxRecord)
79		{
80	6	$outboxRecord['message_id'] = hex2bin($this->stripDashes($outboxRecord['message_id']));
81	6	$outboxRecord['endpoint_id'] = $endpointId;
82	6	$outboxRecord['is_dispatched'] = 0;
83
84	6	$this->connection->transactional(
85	6	function (Connection $connection) use ($outboxRecord) {
86	6	$connection->insert($this->messagesTableName, $outboxRecord);
87	6	}
88		);
89	6	}
90
91		/**
92		* @param int $endpointId
93		* @param string $messageId
94		*
95		* @throws \Throwable
96		*/
97	4	public function markAsDispatched($endpointId, $messageId)
98		{
99		try {
100	4	$this->connection->executeUpdate(
101	4	"UPDATE {$this->messagesTableName}
102		SET is_dispatched = 1, dispatched_at = ?, transport_operations = ''
103		WHERE endpoint_id = ? AND message_id = ? AND is_dispatched = 0",
104		[
105	4	(new \DateTime('now', new \DateTimeZone('UTC')))->format('Y-m-d H:i:s'),
106	4	$endpointId,
107	4	hex2bin($this->stripDashes($messageId))
108		]
109		);
110	1	} catch (\Throwable $t) {
111	1	throw $this->attemptToReconnectPresumedLostConnection($t);
112		}
113	3	}
114
115		/**
116		* Initiates the transaction
117		* Attempts to reconnect once if disconnected.
118		*
119		* @return void
120		* @throws \Throwable
121		*/
122	1	public function beginTransaction()
123		{
124		try {
125	1	$this->connection->beginTransaction();
126	1	} catch (\Throwable $t) {
127	1	throw $this->attemptToReconnectPresumedLostConnection($t);
128		}
129		}
130
131		/**
132		* Commits the transaction
133		*
134		* Does not attempt to reconnect if disconnected because the transaction would be broken anyway.
135		* Reconnection should be done by rollBack.
136		*
137		* @return void
138		* @throws ConnectionException
139		*/
140		public function commit()
141		{
142		$this->connection->commit();
143		}
144
145		/**
146		* Rolls back the transaction.
147		* It makes sure that the connection is in the correct state regardless of what happened before.
148		* Correct state means that connection is not rollback only and does not have a transaction nesting level > 0
149		*
150		* @throws \Throwable
151		*/
152	5	public function rollBack()
153		{
154		try {
155		/**
156		* Roll back all the way as this is supposed to be the top level transaction and we want to reset
157		* the nesting level
158		*/
159	5	$transactionNestingLevel = $this->connection->getTransactionNestingLevel();
160	5	for ($i = 0; $i < $transactionNestingLevel - 1; $i++) {
161	1	$this->connection->rollBack();
162		}
163	5	$this->connection->rollBack();
164	4	} catch (\Throwable $t) {
165	4	$rethrowable = $this->attemptToReconnectPresumedLostConnection($t);
166		/**
167		* If connection is functional we need to make sure the connection is not rollback only.
168		* This can only be achieved by starting a transaction and rolling it back (the "why" is found in
169		* lines 1277-1279 of Doctrine\DBAL\Connection).
170		*/
171	4	if ($rethrowable === $t) {
172	4	$this->connection->beginTransaction();
173	4	$this->connection->rollBack();
174		}
175	4	throw $rethrowable;
176		}
177	1	}
178
179		/**
180		* Attempts to reconnect once if disconnected.
181		*
182		* @param \DateTime $dateTime
183		*
184		* @throws \Exception
185		*/
186	2	public function removeEntriesOlderThan(\DateTime $dateTime)
187		{
188	2	$dateTime->setTimezone(new \DateTimeZone('UTC'));
189	2	$this->connection->executeUpdate(
190	2	"DELETE FROM {$this->messagesTableName} WHERE is_dispatched = 1 AND dispatched_at <= ?",
191	2	[$dateTime->format('Y-m-d H:i:s')]
192		);
193	2	}
194
195		/**
196		* @param string $endpointName
197		*
198		* @return int
199		* @throws \Exception
200		*/
201	2	public function fetchOrGenerateEndpointId($endpointName)
202		{
203	2	$endpointId = 0;
204	2	$this->connection->transactional(
205	2	function (Connection $connection) use ($endpointName, &$endpointId) {
206	2	$lookupHash = md5($endpointName);
207	2	$endpointRecord = $connection->executeQuery(
208	2	"SELECT * FROM {$this->endpointsTableName} WHERE lookup_hash = ?",
209	2	[hex2bin($lookupHash)]
210	2	)->fetch(\PDO::FETCH_ASSOC);
211	2	if (!$endpointRecord) {
212	2	$connection->insert(
213	2	$this->endpointsTableName,
214	2	['lookup_hash' => hex2bin($lookupHash), 'name' => $endpointName]
215		);
216	2	$endpointId = (int)$connection->lastInsertId();
217		} else {
218	1	$endpointId = (int)$endpointRecord['id'];
219		}
220
221	2	}
222		);
223
224	2	return $endpointId;
225		}
226
227		/**
228		* @param string $messageId
229		*
230		* @return string
231		*/
232	9	private function stripDashes($messageId)
233		{
234	9	return str_replace('-', '', $messageId);
235		}
236
237		/**
238		* It attempts to reconnect if connection is non responsive. Failing to reconnect triggers a critical error.
239		* If connection is responsive or successfully reconnected it rethrows, relying on the bus retries
240		* to re-execute everything from the beginning.
241		*
242		* @param \Throwable $t
243		*
244		* @return \Throwable\|CriticalErrorException
245		*/
246	5	private function attemptToReconnectPresumedLostConnection(\Throwable $t)
247		{
248		// presumably, any exception caught here is related to some connection error
249	5	if (!$this->connection->ping()) {
250		// if pinging fails, we try to reconnect
251		try {
252	5	$this->connection->close();
253	5	$this->connection->connect();
254		} catch (\Throwable $t) {
255		// if reconnecting fails, there is no way that the bus can continue to function
256		return new CriticalErrorException("Database connection failed.", 0, $t);
257		}
258		}
259
260	5	return $t;
261		}
262		}
263

phpservicebus / persistence-doctrine2

OutboxPersister::get() A last analyzed 2020-01-22 18:29 UTC

Complexity

Size

Duplication

Code Coverage

Importance

Duplication Side-by-Side

Filter issues like

OutboxPersister::get() A
last analyzed 2020-01-22 18:29 UTC