<commit_before>package org.orienteer.notifications.task;

import com.orientechnologies.orient.core.db.document.ODatabaseDocument;
import com.orientechnologies.orient.core.record.impl.ODocument;
import org.apache.wicket.ThreadContext;
import org.orienteer.core.OrienteerWebApplication;
import org.orienteer.core.dao.DAO;
import org.orienteer.notifications.model.IONotification;
import org.orienteer.notifications.model.IONotificationDAO;
import org.orienteer.notifications.module.ONotificationModule;
import org.orienteer.notifications.repository.ONotificationModuleRepository;
import org.orienteer.notifications.scheduler.ONotificationScheduler;
import org.orienteer.notifications.scheduler.ONotificationTask;
import org.orienteer.notifications.service.INotificationService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import ru.ydn.wicket.wicketorientdb.utils.DBClosure;

import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.stream.Collectors;

/**
 * Task for send notifications
 */
public class ONotificationSendTask extends ONotificationTask {

  private static final Logger LOG = LoggerFactory.getLogger(ONotificationSendTask.class);

  public static final String NAME = "send-notifications";

  public ONotificationSendTask() {
    super(NAME);
  }

  @Override
  public void run() {
    DBClosure.sudoConsumer(db -> {
      try {
        sendNotifications(db);
      } catch (Exception e) {
        LOG.error("Error during {} execution!", getClass().getName(), e);
      } finally {
        rescheduleTask(db);
      }
    });
  }

  private void sendNotifications(ODatabaseDocument db) {
    IONotificationDAO notificationDao = IONotificationDAO.get();

    ONotificationModule.Module module = ONotificationModuleRepository.getModule(db);
    ODocument sentStatus = notificationDao.getSentStatus();
    List<ODocument> notificationsDocs = notificationDao.findNotificationsExceptStatus(sentStatus);

    int executorSize = computeExecutorSize(module.getNotificationsPerWorker(), notificationsDocs);
    int notificationsPerWorker = module.getNotificationsPerWorker();
    List<List<ODocument>> groupedNotifications = groupNotificationsForWorkers(notificationsPerWorker, notificationsDocs);
    ExecutorService executorService = Executors.newFixedThreadPool(executorSize);

    LOG.info("Sending {} notifications...", notificationsDocs.size());
    List<Future<?>> futures = submitTasks(groupedNotifications, executorService);
    waitForComplete(futures);
  }

  private void waitForComplete(List<Future<?>> futures) {<commit_msg>Try to send notifications with status PENDING<commit_after>package org.orienteer.notifications.task;

import com.orientechnologies.orient.core.db.document.ODatabaseDocument;
import com.orientechnologies.orient.core.record.impl.ODocument;
import org.apache.wicket.ThreadContext;
import org.orienteer.core.OrienteerWebApplication;
import org.orienteer.core.dao.DAO;
import org.orienteer.notifications.model.IONotification;
import org.orienteer.notifications.model.IONotificationDAO;
import org.orienteer.notifications.module.ONotificationModule;
import org.orienteer.notifications.repository.ONotificationModuleRepository;
import org.orienteer.notifications.scheduler.ONotificationScheduler;
import org.orienteer.notifications.scheduler.ONotificationTask;
import org.orienteer.notifications.service.INotificationService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import ru.ydn.wicket.wicketorientdb.utils.DBClosure;

import java.util.*;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.stream.Collectors;

/**
 * Task for send notifications
 */
public class ONotificationSendTask extends ONotificationTask {

  private static final Logger LOG = LoggerFactory.getLogger(ONotificationSendTask.class);

  public static final String NAME = "send-notifications";

  public ONotificationSendTask() {
    super(NAME);
  }

  @Override
  public void run() {
    DBClosure.sudoConsumer(db -> {
      try {
        sendNotifications(db);
      } catch (Exception e) {
        LOG.error("Error during {} execution!", getClass().getName(), e);
      } finally {
        rescheduleTask(db);
      }
    });
  }

  private void sendNotifications(ODatabaseDocument db) {
    IONotificationDAO notificationDao = IONotificationDAO.get();

    ONotificationModule.Module module = ONotificationModuleRepository.getModule(db);
    ODocument pendingStatus = notificationDao.getPendingStatus();
    List<ODocument> notificationsDocs = notificationDao.findNotificationsByStatus(pendingStatus);

    int executorSize = computeExecutorSize(module.getNotificationsPerWorker(), notificationsDocs);
    int notificationsPerWorker = module.getNotificationsPerWorker();
    List<List<ODocument>> groupedNotifications = groupNotificationsForWorkers(notificationsPerWorker, notificationsDocs);
    ExecutorService executorService = Executors.newFixedThreadPool(executorSize);

    LOG.info("Sending {} notifications...", notificationsDocs.size());
    List<Future<?>> futures = submitTasks(groupedNotifications, executorService);
    waitForComplete(futures);
  }

  private void waitForComplete(List<Future<?>> futures) {
{'fraction_non_alphanumeric': 0.09691300280636109, 'fraction_numerical': 0.0007483629560336763, 'mean_word_length': 5.741488020176544, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 19, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3115234', 'n_tokens_mistral': 1539, 'n_tokens_neox': 1475, 'n_words': 256}