Index: contrib/splitindex/src/java/org/apache/lucene/index/ScheduledSplitRule.java
===================================================================
--- contrib/splitindex/src/java/org/apache/lucene/index/ScheduledSplitRule.java	(revision 0)
+++ contrib/splitindex/src/java/org/apache/lucene/index/ScheduledSplitRule.java	(revision 0)
@@ -0,0 +1,206 @@
+package org.apache.lucene.index;
+
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You 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.
+ */
+
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.Map;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * The <code>ScheduledSplitRule</code> causes a split to occur in the
+ * {@link SplitPolicy} to which it applies, at regular time periods, with or
+ * without an initial delay.
+ * 
+ * @author Karthick Sankarachary
+ */
+public class ScheduledSplitRule extends SplitRuleAdapter implements SplitRule {
+  private static final long serialVersionUID = -2417666229895435976L;
+  
+  // A configuration option specifying the time period at which to split.
+  public static final String SCHEDULED_SPLIT_RULE_TIME_PERIOD = "scheduled.split.rule.time.period";
+  // A configuration option specifying the time unit for the above time period.
+  public static final String SCHEDULED_SPLIT_RULE_TIME_UNIT = "scheduled.split.rule.time.unit";
+  // A configuration option specifying the initial delay before the first split.
+  public static final String SCHEDULED_SPLIT_RULE_INITIAL_DELAY = "scheduled.split.rule.initial.delay";
+  
+  // The time period for which this scheduled split rule is configured.
+  private transient long period;
+  
+  // The time unit for which this scheduled split rule is configured.
+  private transient TimeUnit unit;
+  
+  // The initial delay for which this scheduled split rule is configured.
+  private transient long initialDelay;
+  
+  // An executer service that knows how to schedule a thread at a fixed rate.
+  private transient ScheduledExecutorService scheduler;
+  
+  /**
+   * Construct a scheduled split rule with no specific configuration options.
+   */
+  public ScheduledSplitRule() {
+    super();
+  }
+  
+  /**
+   * Construct a scheduled split rule with the given configuration options.
+   * 
+   * @param options
+   *          the configuration options.
+   */
+  public ScheduledSplitRule(Map<String,Serializable> options) {
+    super(options);
+  }
+  
+  /**
+   * Default the schedule time period to once every day.
+   */
+  @Override
+  public Map<String,Serializable> getOptionDefaults() {
+    Map<String,Serializable> options = super.getOptionDefaults();
+    options.put(ScheduledSplitRule.SCHEDULED_SPLIT_RULE_INITIAL_DELAY, 0);
+    options.put(ScheduledSplitRule.SCHEDULED_SPLIT_RULE_TIME_PERIOD, 1);
+    options.put(ScheduledSplitRule.SCHEDULED_SPLIT_RULE_TIME_UNIT,
+        TimeUnit.DAYS);
+    return options;
+  }
+  
+  /**
+   * Ensure that the time period and (initial delay) is a positive
+   * (non-negative) number and that the time unit is an instance of
+   * {@link TimeUnit}.
+   */
+  @Override
+  public java.util.Map<String,OptionBound> getOptionBounds() {
+    Map<String,OptionBound> bounds = super.getOptionBounds();
+    bounds.put(ScheduledSplitRule.SCHEDULED_SPLIT_RULE_TIME_PERIOD,
+        new OptionBound() {
+          public boolean isInbounds(String key, Serializable value) {
+            return (value instanceof Long && ((Long) value).longValue() > 0)
+                || (value instanceof Integer && ((Integer) value).intValue() > 0);
+          }
+        });
+    bounds.put(ScheduledSplitRule.SCHEDULED_SPLIT_RULE_INITIAL_DELAY,
+        new OptionBound() {
+          public boolean isInbounds(String key, Serializable value) {
+            return (value instanceof Long && ((Long) value).longValue() >= 0)
+                || (value instanceof Integer && ((Integer) value).intValue() >= 0);
+          }
+        });
+    bounds.put(ScheduledSplitRule.SCHEDULED_SPLIT_RULE_TIME_UNIT,
+        new OptionBound() {
+          public boolean isInbounds(String key, Serializable value) {
+            return value instanceof TimeUnit;
+          }
+        });
+    return bounds;
+  };
+  
+  /**
+   * When the split index is opened, schedule a split to occur at the specified
+   * fixed rate.
+   * 
+   * @param splitPolicy
+   *          the split policy to which this rule applies
+   */
+  public void onOpen(SplitPolicy splitPolicy) {
+    super.onOpen(splitPolicy);
+    try {
+      // Configure this particular scheduled split rule.
+      period = (Long) getOption(
+          ScheduledSplitRule.SCHEDULED_SPLIT_RULE_TIME_PERIOD, Long.class);
+      unit = (TimeUnit) options
+          .get(ScheduledSplitRule.SCHEDULED_SPLIT_RULE_TIME_UNIT);
+      initialDelay = (Long) getOption(
+          ScheduledSplitRule.SCHEDULED_SPLIT_RULE_INITIAL_DELAY, Long.class);
+      
+      // Use sensible defaults for the keys in the context.
+      context.put(SplitRule.SPLIT_RULE_SPLIT_POLICY_ID, getSplitPolicyId());
+      context.put(SplitRule.SPLIT_RULE_LAST_SPLIT_TIME, 0L);
+      context.put(SplitRule.SPLIT_RULE_LAST_SEGMENT_NAME, null);
+      context.put(SplitRule.SPLIT_RULE_LAST_SEGMENT_SIZE, 0L);
+      
+      // Let the method below schedule the split.
+      scheduler = Executors.newSingleThreadScheduledExecutor();
+      scheduler.scheduleAtFixedRate(new Runnable() {
+        public void run() {
+          if (!paused) {
+            try {
+              ScheduledSplitRule.this.splitPolicy.split(context);
+            } catch (Exception e) {
+              System.out.println("The scheduled split failed ("
+                  + e.getMessage()
+                  + "), hence putting the split rule on pause.");
+              e.printStackTrace();
+              onPause();
+            }
+          }
+        }
+        
+      }, initialDelay, period, unit);
+      
+    } catch (Exception e) {
+      throw new RuntimeException(e);
+    }
+  }
+  
+  /**
+   * When the split index is closed, shutdown the scheduler.
+   */
+  public SplitVote onClose() throws CorruptIndexException, SplitException,
+      IOException {
+    scheduler.shutdown();
+    return super.onClose();
+  }
+  
+  /**
+   * When the split writer is paused, simply shutdown the scheduler.
+   */
+  @Override
+  public synchronized SplitVote onPause() {
+    scheduler.shutdown();
+    return super.onPause();
+  }
+  
+  /**
+   * When the split writer is resumed, schedule the split all over again.
+   */
+  @Override
+  public synchronized SplitVote onResume() {
+    scheduler = Executors.newSingleThreadScheduledExecutor();
+    scheduler.scheduleAtFixedRate(new Runnable() {
+      public void run() {
+        if (!paused) {
+          try {
+            ScheduledSplitRule.this.splitPolicy.split(context);
+          } catch (Exception e) {
+            System.out.println("The scheduled split failed (" + e.getMessage()
+                + "), hence putting the split rule on pause.");
+            e.printStackTrace();
+            onPause();
+          }
+        }
+      }
+      
+    }, initialDelay, period, unit);
+    return super.onResume();
+  }
+}