src/main/java/com/googlesource/gerrit/plugins/rabbitmq/message/MessagePublisher.java

// Copyright (C) 2015 The Android Open Source Project
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package com.googlesource.gerrit.plugins.rabbitmq.message;

import com.google.gerrit.common.ChangeHooks;
import com.google.gerrit.common.ChangeListener;
import com.google.gerrit.extensions.events.LifecycleListener;
import com.google.gerrit.reviewdb.client.Account;
import com.google.gerrit.reviewdb.server.ReviewDb;
import com.google.gerrit.server.CurrentUser;
import com.google.gerrit.server.IdentifiedUser;
import com.google.gerrit.server.PluginUser;
import com.google.gerrit.server.account.AccountResolver;
import com.google.gerrit.server.events.ChangeEvent;
import com.google.gerrit.server.git.WorkQueue;
import com.google.gerrit.server.util.RequestContext;
import com.google.gerrit.server.util.ThreadLocalRequestContext;
import com.google.gson.Gson;
import com.google.gwtorm.server.OrmException;
import com.google.gwtorm.server.SchemaFactory;
import com.google.inject.Inject;
import com.google.inject.Provider;
import com.google.inject.ProvisionException;
import com.google.inject.assistedinject.Assisted;

import com.googlesource.gerrit.plugins.rabbitmq.Keys;
import com.googlesource.gerrit.plugins.rabbitmq.session.AMQPSession;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Timer;
import java.util.TimerTask;

public class MessagePublisher implements ChangeListener, LifecycleListener {

  public interface Factory {
    MessagePublisher create(AMQPSession session);
  }

  private static final Logger LOGGER = LoggerFactory.getLogger(MessagePublisher.class);

  private final static int MONITOR_FIRSTTIME_DELAY = 15000;

  private final AMQPSession session;
  private final ChangeHooks hooks;
  private final Gson gson;
  private final WorkQueue workQueue;
  private final AccountResolver accountResolver;
  private final IdentifiedUser.GenericFactory userFactory;
  private final ThreadLocalRequestContext threadLocalRequestContext;
  private final PluginUser pluginUser;
  private final SchemaFactory<ReviewDb> schemaFactory;
  private final Timer monitorTimer = new Timer();
  private ReviewDb db;
  private Account userAccount;

  @Inject
  public MessagePublisher(
      @Assisted AMQPSession session,
      ChangeHooks hooks,
      Gson gson,
      WorkQueue workQueue,
      AccountResolver accountResolver,
      IdentifiedUser.GenericFactory userFactory,
      ThreadLocalRequestContext threadLocalRequestContext,
      PluginUser pluginUser,
      SchemaFactory<ReviewDb> schemaFactory) {
    this.session = session;
    this.hooks = hooks;
    this.gson = gson;
    this.workQueue = workQueue;
    this.accountResolver = accountResolver;
    this.userFactory = userFactory;
    this.threadLocalRequestContext = threadLocalRequestContext;
    this.pluginUser = pluginUser;
    this.schemaFactory = schemaFactory;
  }

  @Override
  public void start() {
    LOGGER.info("Start identified listener.");
    session.connect();
    monitorTimer.schedule(new TimerTask() {
      @Override
      public void run() {
        if (!session.isOpen()) {
          LOGGER.info("#start: try to reconnect");
          session.connect();
        }
      }
    }, MONITOR_FIRSTTIME_DELAY, session.getProperties().getInt(Keys.MONITOR_INTERVAL));

    if (session.getProperties().hasListenAs()) {
      final String userName = session.getProperties().getListenAs();
      final ChangeListener changeListener = this;
      workQueue.getDefaultQueue().submit(new Runnable() {
        @Override
        public void run() {
          RequestContext old = threadLocalRequestContext
              .setContext(new RequestContext() {

                @Override
                public CurrentUser getCurrentUser() {
                  return pluginUser;
                }

                @Override
                public Provider<ReviewDb> getReviewDbProvider() {
                  return new Provider<ReviewDb>() {
                    @Override
                    public ReviewDb get() {
                      if (db == null) {
                        try {
                          db = schemaFactory.open();
                        } catch (OrmException e) {
                          throw new ProvisionException("Cannot open ReviewDb", e);
                        }
                      }
                      return db;
                    }
                  };
                }
              });
          try {
            userAccount = accountResolver.find(userName);
            if (userAccount == null) {
              LOGGER.error("No single user could be found when searching for listenAs: {}", userName);
              return;
            }

            IdentifiedUser user = userFactory.create(userAccount.getId());
            hooks.addChangeListener(changeListener, user);
            LOGGER.info("Listen events as : {}", userName);
          } catch (OrmException e) {
            LOGGER.error("Could not query database for listenAs", e);
            return;
          } finally {
            threadLocalRequestContext.setContext(old);
            if (db != null) {
              db.close();
              db = null;
            }
          }
        }
      });
    }
  }

  @Override
  public void stop() {
    LOGGER.info("Start identified listener.");
    monitorTimer.cancel();
    session.disconnect();
    hooks.removeChangeListener(this);
  }

  @Override
  public void onChangeEvent(ChangeEvent event) {
    if (session.isOpen()) {
      session.publishMessage(gson.toJson(event));
    }
  }
}