<commit_before>package io.kubernetes.client.informer.cache;

import io.kubernetes.client.informer.ResourceEventHandler;
import io.kubernetes.client.informer.exception.BadNotificationException;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import org.joda.time.DateTime;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * ProcessorListener implements Runnable interface. It's supposed to run in background and actually
 * executes its event handler on notification. Note that it allows 1000 pending notifications at
 * maximum.
 */
public class ProcessorListener<ApiType> implements Runnable {

  private static final Logger log = LoggerFactory.getLogger(ProcessorListener.class);

  private static final int DEFAULT_QUEUE_CAPACITY = 1000;

  // resyncPeriod is how frequently the listener wants a full resync from the shared informer. This
  // value may differ from requestedResyncPeriod if the shared informer adjusts it to align with the
  // informer's overall resync check period.
  private long resyncPeriod;
  private DateTime nextResync;

  private BlockingQueue<Notification> queue;

  private ResourceEventHandler<ApiType> handler;

  public ProcessorListener(ResourceEventHandler<ApiType> handler, long resyncPeriod) {
    this.resyncPeriod = resyncPeriod;
    this.handler = handler;

    this.queue = new ArrayBlockingQueue<>(DEFAULT_QUEUE_CAPACITY);

    determineNextResync(DateTime.now());
  }

  @Override
  public void run() {
    while (true) {
      try {
        Notification obj = queue.take();
        if (obj instanceof UpdateNotification) {
          UpdateNotification notification = (UpdateNotification) obj;
          this.handler.onUpdate(
              (ApiType) notification.getOldObj(), (ApiType) notification.getNewObj());
        } else if (obj instanceof AddNotification) {
          AddNotification notification = (AddNotification) obj;
          this.handler.onAdd((ApiType) notification.getNewObj());
        } else if (obj instanceof DeleteNotification) {
          Object deletedObj = ((DeleteNotification) obj).getOldObj();
          if (deletedObj instanceof DeltaFIFO.DeletedFinalStateUnknown) {
            this.handler.onDelete(
                ((DeltaFIFO.DeletedFinalStateUnknown<ApiType>) deletedObj).getObj(), true);
          } else {
            this.handler.onDelete((ApiType) deletedObj, false);
          }
        } else {
          throw new BadNotificationException("unrecognized notification");
        }
      } catch (InterruptedException e) {
        log.error("processor interrupted: {}", e.getMessage());
        return;
      }
    }
  }

  public void add(Notification<ApiType> obj) {
    if (obj == null) {<commit_msg>fixes listener quiting unexpectedly<commit_after>package io.kubernetes.client.informer.cache;

import io.kubernetes.client.informer.ResourceEventHandler;
import io.kubernetes.client.informer.exception.BadNotificationException;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.BlockingQueue;
import org.joda.time.DateTime;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * ProcessorListener implements Runnable interface. It's supposed to run in background and actually
 * executes its event handler on notification. Note that it allows 1000 pending notifications at
 * maximum.
 */
public class ProcessorListener<ApiType> implements Runnable {

  private static final Logger log = LoggerFactory.getLogger(ProcessorListener.class);

  private static final int DEFAULT_QUEUE_CAPACITY = 1000;

  // resyncPeriod is how frequently the listener wants a full resync from the shared informer. This
  // value may differ from requestedResyncPeriod if the shared informer adjusts it to align with the
  // informer's overall resync check period.
  private long resyncPeriod;
  private DateTime nextResync;

  private BlockingQueue<Notification> queue;

  private ResourceEventHandler<ApiType> handler;

  public ProcessorListener(ResourceEventHandler<ApiType> handler, long resyncPeriod) {
    this.resyncPeriod = resyncPeriod;
    this.handler = handler;

    this.queue = new ArrayBlockingQueue<>(DEFAULT_QUEUE_CAPACITY);

    determineNextResync(DateTime.now());
  }

  @Override
  public void run() {
    while (true) {
      try {
        Notification obj = queue.take();
        if (obj instanceof UpdateNotification) {
          UpdateNotification notification = (UpdateNotification) obj;
          try {
            this.handler.onUpdate(
                (ApiType) notification.getOldObj(), (ApiType) notification.getNewObj());
          } catch (Throwable t) {
            // Catch all exceptions here so that listeners won't quit unexpectedly
            log.error("failed invoking UPDATE event handler: {}", t.getMessage());
            continue;
          }
        } else if (obj instanceof AddNotification) {
          AddNotification notification = (AddNotification) obj;
          try {
            this.handler.onAdd((ApiType) notification.getNewObj());
          } catch (Throwable t) {
            // Catch all exceptions here so that listeners won't quit unexpectedly
            log.error("failed invoking ADD event handler: {}", t.getMessage());
            continue;
          }
        } else if (obj instanceof DeleteNotification) {
          Object deletedObj = ((DeleteNotification) obj).getOldObj();
          try {
            if (deletedObj instanceof DeltaFIFO.DeletedFinalStateUnknown) {
              this.handler.onDelete(
                  ((DeltaFIFO.DeletedFinalStateUnknown<ApiType>) deletedObj).getObj(), true);
            } else {
              this.handler.onDelete((ApiType) deletedObj, false);
            }
          } catch (Throwable t) {
            // Catch all exceptions here so that listeners won't quit unexpectedly
            log.error("failed invoking DELETE event handler: {}", t.getMessage());
            continue;
          }
        } else {
          throw new BadNotificationException("unrecognized notification");
        }
      } catch (InterruptedException e) {
        log.error("processor interrupted: {}", e.getMessage());
        return;
      }
    }
  }

  public void add(Notification<ApiType> obj) {
    if (obj == null) {
{'fraction_non_alphanumeric': 0.08751793400286945, 'fraction_numerical': 0.003188267176789415, 'mean_word_length': 3.1467283542630535, 'pattern_counts': {'":': 0, '<': 17, '<?xml version=': 0, '>': 17, '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': '3231222', 'n_tokens_mistral': 1613, 'n_tokens_neox': 1567, 'n_words': 454}