3 namespace Wallabag\ImportBundle\Import
;
5 use Psr\Log\LoggerInterface
;
6 use Psr\Log\NullLogger
;
7 use Doctrine\ORM\EntityManager
;
8 use Wallabag\CoreBundle\Helper\ContentProxy
;
9 use Wallabag\CoreBundle\Entity\Entry
;
10 use Wallabag\CoreBundle\Entity\Tag
;
11 use Wallabag\UserBundle\Entity\User
;
12 use OldSound\RabbitMqBundle\RabbitMq\ProducerInterface
;
13 use Symfony\Component\EventDispatcher\EventDispatcherInterface
;
14 use Wallabag\CoreBundle\Event\EntrySavedEvent
;
16 abstract class AbstractImport
implements ImportInterface
20 protected $contentProxy;
21 protected $eventDispatcher;
24 protected $markAsRead;
25 protected $skippedEntries = 0;
26 protected $importedEntries = 0;
27 protected $queuedEntries = 0;
29 public function __construct(EntityManager
$em, ContentProxy
$contentProxy, EventDispatcherInterface
$eventDispatcher)
32 $this->logger
= new NullLogger();
33 $this->contentProxy
= $contentProxy;
34 $this->eventDispatcher
= $eventDispatcher;
37 public function setLogger(LoggerInterface
$logger)
39 $this->logger
= $logger;
43 * Set RabbitMQ/Redis Producer to send each entry to a queue.
44 * This method should be called when user has enabled RabbitMQ.
46 * @param ProducerInterface $producer
48 public function setProducer(ProducerInterface
$producer)
50 $this->producer
= $producer;
55 * Could the current *connected* user or one retrieve by the consumer.
59 public function setUser(User
$user)
65 * Set whether articles must be all marked as read.
67 * @param bool $markAsRead
69 public function setMarkAsRead($markAsRead)
71 $this->markAsRead
= $markAsRead;
77 * Get whether articles must be all marked as read.
79 public function getMarkAsRead()
81 return $this->markAsRead
;
85 * Fetch content from the ContentProxy (using graby).
86 * If it fails return the given entry to be saved in all case (to avoid user to loose the content).
88 * @param Entry $entry Entry to update
89 * @param string $url Url to grab content for
90 * @param array $content An array with AT LEAST keys title, html, url, language & content_type to skip the fetchContent from the url
94 protected function fetchContent(Entry
$entry, $url, array $content = [])
97 return $this->contentProxy
->updateEntry($entry, $url, $content);
98 } catch (\Exception
$e) {
104 * Parse and insert all given entries.
108 protected function parseEntries($entries)
111 $entryToBeFlushed = [];
113 foreach ($entries as $importedEntry) {
114 if ($this->markAsRead
) {
115 $importedEntry = $this->setEntryAsRead($importedEntry);
118 $entry = $this->parseEntry($importedEntry);
120 if (null === $entry) {
124 // store each entry to be flushed so we can trigger the entry.saved event for each of them
125 // entry.saved needs the entry to be persisted in db because it needs it id to generate
127 $entryToBeFlushed[] = $entry;
129 // flush every 20 entries
130 if (($i %
20) === 0) {
133 foreach ($entryToBeFlushed as $entry) {
134 $this->eventDispatcher
->dispatch(EntrySavedEvent
::NAME
, new EntrySavedEvent($entry));
137 $entryToBeFlushed = [];
139 // clear only affected entities
140 $this->em
->clear(Entry
::class);
141 $this->em
->clear(Tag
::class);
148 if (!empty($entryToBeFlushed)) {
149 foreach ($entryToBeFlushed as $entry) {
150 $this->eventDispatcher
->dispatch(EntrySavedEvent
::NAME
, new EntrySavedEvent($entry));
156 * Parse entries and send them to the queue.
157 * It should just be a simple loop on all item, no call to the database should be done
158 * to speedup queuing.
160 * Faster parse entries for Producer.
161 * We don't care to make check at this time. They'll be done by the consumer.
163 * @param array $entries
165 protected function parseEntriesForProducer(array $entries)
167 foreach ($entries as $importedEntry) {
168 // set userId for the producer (it won't know which user is connected)
169 $importedEntry['userId'] = $this->user
->getId();
171 if ($this->markAsRead
) {
172 $importedEntry = $this->setEntryAsRead($importedEntry);
175 ++
$this->queuedEntries
;
177 $this->producer
->publish(json_encode($importedEntry));
184 public function getSummary()
187 'skipped' => $this->skippedEntries
,
188 'imported' => $this->importedEntries
,
189 'queued' => $this->queuedEntries
,
196 * @param array $importedEntry
200 abstract public function parseEntry(array $importedEntry);
203 * Set current imported entry to archived / read.
204 * Implementation is different accross all imports.
206 * @param array $importedEntry
210 abstract protected function setEntryAsRead(array $importedEntry);