在未读取所有元素时清理 Iterable

Question

开始接触 RxJava。我有一个实现 Iterable 的 class 我想转换为 Observable。使用 Observable.from() 似乎很容易。但是，我需要设置和拆除为我提供各个条目的代码（迭代器中的 next()。

当我运行完成整个序列时，这很容易。我添加了对 hasNext() 函数的调用，当没有下一个时，我运行进行了拆解。然而，我想使用的非常有前途的运算符之一是 take(someNumber)。如果在 Iterator 运行s 之前停止获取项目，则清理代码永远不会运行s.

我可以做些什么来清理运行ning？如果使用 from(Iterable) 以外的其他内容，我可以接受。我现在卡在 Java6 上了。为了说明我的困境，我创建了一个最小样本：

更新：基于不要将 Iterator 和 Iterable 混合在一起的反馈，我更新了下面的代码。要理解原始答案，original code is in that gist.

更新的测试代码（仍然不好）：

import rx.Observable;
import rx.functions.Action0;
import rx.functions.Action1;

/**
* @author stw
*
*/
public class RXTest {

/**
 * @param args
 */
public static void main(String[] args) {
  ComplicatedObject co = new ComplicatedObject();
  Observable<FancyObject> fancy = Observable.from(co);
  // if the take is less than the elements cleanup never
  // runs. If you take the take out, cleanup runs
  fancy.take(3).subscribe(
      new Action1<FancyObject>() {

        public void call(FancyObject item) {
            System.out.println(item.getName());
        }
    },
    new Action1<Throwable>() {

        public void call(Throwable error) {
            System.out.println("Error encountered: " + error.getMessage());
        }
    },
    new Action0() {

        public void call() {
            System.out.println("Sequence complete");
        }
    }

      );

}

}

花哨的对象：

import java.util.Date;
import java.util.UUID;

/**
* @author stw
*
*/
public class FancyObject  {
private String name = UUID.randomUUID().toString();
private Date created = new Date();
public String getName() {
  return this.name;
}
public void setName(String name) {
  this.name = name;
}
public Date getCreated() {
  return this.created;
}
public void setCreated(Date created) {
  this.created = created;
}
}

迭代器：

import java.util.Iterator;

/**
 * @author stw
 *
 */
public class FancyIterator implements Iterator<FancyObject> {

  private final ComplicatedObject theObject;
  private int fancyCount = 0;


  public FancyIterator(ComplicatedObject co) {
    this.theObject = co;
  }

  public boolean hasNext() {
    return this.theObject.hasObject(this.fancyCount);
   }


   public FancyObject next() {
     FancyObject result = this.theObject.getOne(this.fancyCount);
     this.fancyCount++;  
     return result;
   }

}

可迭代对象：

import java.util.Iterator;
import java.util.Vector;

/**
 * @author stw
 *
 */
public class ComplicatedObject implements Iterable<FancyObject> {

  private boolean isInitialized = false;

  Vector<FancyObject> allOfThem = new Vector<FancyObject>();


  public Iterator<FancyObject> iterator() {
   return new FancyIterator(this);
  }

  public boolean hasObject(int whichone) {
    if (!this.isInitialized) {
      this.setupAccesstoFancyObject(); 
    }
    return (whichone < this.allOfThem.size());
  }
  public FancyObject getOne(int whichone) {
      if (!this.isInitialized) {
        this.setupAccesstoFancyObject();
      }
      if (whichone < this.allOfThem.size()) {
        return this.allOfThem.get(whichone);
      }
      // If we ask bejond...
      this.isInitialized = false;
      this.teardownAccessToFancyObjects();
      return null;
  }

  private void setupAccesstoFancyObject() {
    System.out.println("Initializing fancy objects");
    for (int i = 0; i < 20; i++) {
      this.allOfThem.addElement(new FancyObject());
    }
    this.isInitialized = true;
  }

  private void teardownAccessToFancyObjects() {
    System.out.println("I'm doing proper cleanup here");

  }

}

但是真正的问题（谢谢@Andreas）似乎是：

当底层代码需要 setup/teardown 时，我可以使用什么构造来创建 Observable，尤其是当人们期望并非所有元素都被拉出时。 Iterable 只是我的第一个想法

更新 2：根据 Dave 的回答，我 created a gist 使用了我的工作解决方案。迭代器并不完美，但它是一个开始。

Answer 1

如果您想要那种控制，您需要将 Iterable 的实现与 Iterator 分开。 Iterable 意味着 class 可以提供对 class 有意义的任何形式的 Iterator。

但是，如果您在同一个 class 中实现 Iterator，那么您将只能为每个 ComplicatedObject 实例设置一个 Iterator。正确的做法是实施

class FancyObjectIterator implements Iterator<FancyObject>
{
    ...
}

与 ComplicatedObject 分开，因此您可以在使用完后仅丢弃 partially-used 迭代器。 ComplicatedObject 应该只实施 Iterable<FancyObject>.

如果您 object 由于迭代器具有更多需要特殊清理的状态而采用该方法，那么您的设计有问题。 Iterator 应该知道的唯一状态是基 "collection" 中的当前位置，对于 "collection" 和 "position" 的非常宽松的定义，因为迭代器的概念可以适用于远远超过典型的 collections.

Answer 2

您不能同时实施 Iterator 和 Iterable，因为 Iterable.iterator() 必须 return 一个 new Iterator 或每次调用。

允许代码并行地多次迭代相同的 Iterable。

示例：在 Iterable:

中查找重复元素的一种过于简化的方法

Iterable<MyObject> myIterable = ...;
for (MyObject myObj1 : myIterable) {
    for (MyObject myObj2 : myIterable) {
        if (myObj1 != myObj2 && myObj1.equals(myObj2)) {
            // found duplicate
        }
    }
}

此处使用的增强型 for 循环将分别使用 Iterator.

如您所见，每个Iterator都必须保持自己的独立地位。因此，iterator() 方法需要 return 一个具有自己状态的新对象。

对于您关于清理代码的问题，Iterator 没有 close() 方法。迭代器状态不应该需要清理。如果他们绝对必须，终结器可以做到这一点，但终结器可能需要非常很长时间才能被调用。终结器的一般建议是：不要。

Answer 3

Observable.using 用于终止（完成或错误）或取消订阅。要使用它，您需要使拆卸代码可访问，以便您的源可观察对象看起来像这样：

source = Observable.using(
    resourceFactory, 
    observableFactory, 
    resourceDisposer);

您的代码可能如下所示：

source = Observable.using(
    () -> new ComplicatedObject(),
    co -> Observable.from(co), 
    co -> co.tearDown());

在未读取所有元素时清理 Iterable

Cleaning up an Iterable when not all elements are read

java

reactive-programming

java-6

rx-java