package com.celements.common.observation.listener;

import static java.text.MessageFormat.*;

import java.io.NotSerializableException;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.xwiki.component.annotation.Component;
import org.xwiki.observation.event.Event;
import org.xwiki.observation.remote.RemoteObservationManager;

import com.celements.common.observation.converter.Local;

/**
 * Register to {@link org.xwiki.observation.ObservationManager} for all non-local events
 * and send them to {@link RemoteObservationManager}.
 */
@Component(LocalEventListener.NAME)
public class LocalEventListener extends org.xwiki.observation.remote.internal.LocalEventListener {

  private Logger LOGGER = LoggerFactory.getLogger(LocalEventListener.class);

  public static final String NAME = "observation.remote";

  final Map<Class<? extends Event>, LogCounter> logCountMap = new ConcurrentHashMap<>();

  @Override
  public String getName() {
    return NAME;
  }

  @Override
  public void onEvent(Event event, Object source, Object data) {
    Class<? extends Event> type = event.getClass();
    if (type.isAnnotationPresent(Local.class)) {
      LOGGER.info("skipping local event '{}'", event.getClass());
    } else {
      try {
        super.onEvent(event, source, data);
      } catch (Exception exc) {
        if (containsNotSerializableException(exc)) {
          logNotSerializableException(type, source, data, exc);
        } else {
          LOGGER.error("unable to notify remote event [{}], source [{}], data [{}]", type, source,
              data, exc);
        }
      }
    }
  }

  private boolean containsNotSerializableException(Throwable exc) {
    return (exc != null) && ((exc instanceof NotSerializableException)
        || containsNotSerializableException(exc.getCause()));
  }

  private void logNotSerializableException(Class<? extends Event> type, Object source, Object data,
      Exception exc) {
    String msg = "not serializable remote event [{0}], source [{1}], data [{2}]";
    LogCounter logCount = logCountMap.get(type);
    if ((logCount == null) || logCount.isOneHourAgo()) {
      logCountMap.put(type, new LogCounter());
      if (logCount != null) {
        msg += ", occured {3} times within last hour (since {4})";
        msg = format(msg, type.getSimpleName(), source, data, logCount.count, logCount.time);
      } else {
        msg = format(msg, type.getSimpleName(), source, data);
      }
      LOGGER.warn(msg, exc);
    } else if (LOGGER.isDebugEnabled()) {
      LOGGER.debug(format(msg, type.getSimpleName(), source, data), exc);
    }
    logCountMap.get(type).increment();
  }

  static class LogCounter {

    final long time = System.currentTimeMillis();
    final AtomicLong count = new AtomicLong();

    public boolean isOneHourAgo() {
      return (System.currentTimeMillis() - time) >= (1000L * 60 * 60);
    }

    public void increment() {
      count.incrementAndGet();
    }

  }

  /**
   * for test purposes only
   */
  Logger injectLogger(Logger logger) {
    LOGGER = logger;
    return LOGGER;
  }

}

@@ -2142,115 +2142,127 @@
     
-L
+if (!l
 ogCount
-er logCount = logCountMap.get(type);%0A    if ((logCount == null) %7C%7C logCount.isOneHourAgo()) %7B%0A     
+Map.containsKey(type) %7C%7C logCountMap.get(type).isOneHourAgo()) %7B%0A      final LogCounter replacedLogCount =
  log
@@ -2309,17 +2309,25 @@
     if (
-l
+replacedL
 ogCount 
@@ -2466,17 +2466,25 @@
 , data, 
-l
+replacedL
 ogCount.
@@ -2489,18 +2489,38 @@
 t.count,
- l
+%0A            replacedL
 ogCount.

improve logCount (#162)