Line data    Source code

       1             : // Copyright (C) 2016 The Android Open Source Project
       2             : //
       3             : // Licensed under the Apache License, Version 2.0 (the "License");
       4             : // you may not use this file except in compliance with the License.
       5             : // You may obtain a copy of the License at
       6             : //
       7             : // http://www.apache.org/licenses/LICENSE-2.0
       8             : //
       9             : // Unless required by applicable law or agreed to in writing, software
      10             : // distributed under the License is distributed on an "AS IS" BASIS,
      11             : // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
      12             : // See the License for the specific language governing permissions and
      13             : // limitations under the License.
      14             : 
      15             : package com.google.gerrit.server.mail.receive;
      16             : 
      17             : import com.google.common.annotations.VisibleForTesting;
      18             : import com.google.common.flogger.FluentLogger;
      19             : import com.google.gerrit.extensions.events.LifecycleListener;
      20             : import com.google.gerrit.extensions.restapi.RestApiException;
      21             : import com.google.gerrit.lifecycle.LifecycleModule;
      22             : import com.google.gerrit.mail.MailMessage;
      23             : import com.google.gerrit.server.git.WorkQueue;
      24             : import com.google.gerrit.server.mail.EmailSettings;
      25             : import com.google.gerrit.server.update.UpdateException;
      26             : import com.google.inject.Inject;
      27             : import java.io.IOException;
      28             : import java.util.Collections;
      29             : import java.util.HashSet;
      30             : import java.util.List;
      31             : import java.util.Set;
      32             : import java.util.Timer;
      33             : import java.util.TimerTask;
      34             : import java.util.concurrent.Future;
      35             : 
      36             : /** MailReceiver implements base functionality for receiving emails. */
      37             : public abstract class MailReceiver implements LifecycleListener {
      38           2 :   private static final FluentLogger logger = FluentLogger.forEnclosingClass();
      39             : 
      40             :   protected EmailSettings mailSettings;
      41             :   protected Set<String> pendingDeletion;
      42             :   private MailProcessor mailProcessor;
      43             :   private WorkQueue workQueue;
      44             :   private Timer timer;
      45             : 
      46             :   public static class MailReceiverModule extends LifecycleModule {
      47             :     private final EmailSettings mailSettings;
      48             : 
      49             :     @Inject
      50         138 :     MailReceiverModule(EmailSettings mailSettings) {
      51         138 :       this.mailSettings = mailSettings;
      52         138 :     }
      53             : 
      54             :     @Override
      55             :     protected void configure() {
      56         138 :       if (mailSettings.protocol == Protocol.NONE) {
      57         137 :         return;
      58             :       }
      59           2 :       listener().to(MailReceiver.class);
      60           2 :       switch (mailSettings.protocol) {
      61             :         case IMAP:
      62           1 :           bind(MailReceiver.class).to(ImapMailReceiver.class);
      63           1 :           break;
      64             :         case POP3:
      65           2 :           bind(MailReceiver.class).to(Pop3MailReceiver.class);
      66           2 :           break;
      67             :         case NONE:
      68             :         default:
      69             :       }
      70           2 :     }
      71             :   }
      72             : 
      73           2 :   MailReceiver(EmailSettings mailSettings, MailProcessor mailProcessor, WorkQueue workQueue) {
      74           2 :     this.mailSettings = mailSettings;
      75           2 :     this.mailProcessor = mailProcessor;
      76           2 :     this.workQueue = workQueue;
      77           2 :     pendingDeletion = Collections.synchronizedSet(new HashSet<>());
      78           2 :   }
      79             : 
      80             :   @Override
      81             :   public void start() {
      82           2 :     if (timer == null) {
      83           2 :       timer = new Timer();
      84             :     } else {
      85           0 :       timer.cancel();
      86             :     }
      87           2 :     timer.scheduleAtFixedRate(
      88           2 :         new TimerTask() {
      89             :           @Override
      90             :           public void run() {
      91             :             try {
      92           0 :               MailReceiver.this.handleEmails(true);
      93           2 :             } catch (MailTransferException | IOException e) {
      94           2 :               logger.atSevere().withCause(e).log("Error while fetching emails");
      95           0 :             }
      96           2 :           }
      97             :         },
      98             :         0L,
      99             :         mailSettings.fetchInterval);
     100           2 :   }
     101             : 
     102             :   @Override
     103             :   public void stop() {
     104           2 :     if (timer != null) {
     105           2 :       timer.cancel();
     106             :     }
     107           2 :   }
     108             : 
     109             :   /**
     110             :    * requestDeletion will enqueue an email for deletion and delete it the next time we connect to
     111             :    * the email server. This does not guarantee deletion as the Gerrit instance might fail before we
     112             :    * connect to the email server.
     113             :    */
     114             :   public void requestDeletion(String messageId) {
     115           1 :     pendingDeletion.add(messageId);
     116           1 :   }
     117             : 
     118             :   /**
     119             :    * handleEmails will open a connection to the mail server, remove emails where deletion is
     120             :    * pending, read new email and close the connection.
     121             :    *
     122             :    * @param async determines if processing messages should happen asynchronously
     123             :    * @throws MailTransferException in case of a known transport failure
     124             :    * @throws IOException in case of a low-level transport failure
     125             :    */
     126             :   @VisibleForTesting
     127             :   public abstract void handleEmails(boolean async) throws MailTransferException, IOException;
     128             : 
     129             :   protected void dispatchMailProcessor(List<MailMessage> messages, boolean async) {
     130           1 :     for (MailMessage m : messages) {
     131           1 :       if (async) {
     132             :         @SuppressWarnings("unused")
     133           0 :         Future<?> possiblyIgnoredError =
     134             :             workQueue
     135           0 :                 .getDefaultQueue()
     136           0 :                 .submit(
     137             :                     () -> {
     138             :                       try {
     139           0 :                         mailProcessor.process(m);
     140           0 :                         requestDeletion(m.id());
     141           0 :                       } catch (RestApiException | UpdateException e) {
     142           0 :                         logger.atSevere().withCause(e).log(
     143           0 :                             "Mail: Can't process message %s . Won't delete.", m.id());
     144           0 :                       }
     145           0 :                     });
     146           0 :       } else {
     147             :         // Synchronous processing is used only in tests.
     148             :         try {
     149           1 :           mailProcessor.process(m);
     150           1 :           requestDeletion(m.id());
     151           0 :         } catch (RestApiException | UpdateException e) {
     152           0 :           logger.atSevere().withCause(e).log("Mail: Can't process messages. Won't delete.");
     153           1 :         }
     154             :       }
     155           1 :     }
     156           1 :   }
     157             : }