Design Pattern for Java: Does this look familiar to anyone?

Question

In my application, I have the following recurring pattern where I get the feeling that this is an incomplete or warped version of some common pattern that I'm not aware of. In my implementation, I can't seem to find a good way to implement a particular operation, as described below.

A collection of objects, let's call them "sources", produces data if you call a method on them. An instance of ScheduledThreadPoolExecutor (from java.util.concurrent) is used to periodically do readings on the sources and put the data on a common outbound queue:

--o
      method calls    +-----+
--o       <--         |  o  |     --->    |||||||||||
          -->         | o o |
--o      data         +-----+

sources             thread pool          outbound queue

The sources, the thread pool and the queue are members of a class that provides methods to add sources and to retrieve the head of the queue.

Below is the code to add a new source. I'm wondering what happens if I remove a source from the collection.

public void vAddSource(ISource src)
{
    this._rgSources.add(src);
    final ISource srcT = src;
    // q is the outbound queue, which is a private member in the class
    // defining this method. Name is abbreviated for readability
    Runnable cmd = new Runnable() { public void run() { q.add(srcT.sRead()); } }
    // tpx is the thread pool
    this._tpx.scheduleAtFixedRate(cmd, 0L, 1000L, TimeUnit.MILLISECONDS);
}

Now, the way Java works - if I got that correctly -, src etc. are object identifiers that get passed around by value, so in the above code there is only one ISource object and an identifier for it gets added to the collection as well as passed into the thread pool. If I remove the identifier from the collection, the object is still there, hence the thread pool will continue to read from it as if nothing happened.

Am I understanding this correctly? And what would be a typical approach if you wanted the thread pool to take note of the removal and also discard the source?

Answer 1

ThreadPoolExecutor (and hence ScheduledThreadPoolExecutor) has a remove(Runnable) method. You might also be interested in FutureTask which can be cancelled (although that wont take it out of the work queue).

ETA (by Hanno Fietz, from pgras' answer): This also requires to map the Runnables to a source so that the removeSource method could call ThreadPoolExecutor.remove with the right one.

Answer 2

One solution would be to keep a mapping between sources and Runnables so you could implement a removeSource method.

Another way would be to keep a WeakReference to the source instead of a hard one and to let the Runnable terminate itself if the source was garbage collected.