org.simantics.district.maps/src/org/simantics/maps/pojo/TileJob.java

   1 /*******************************************************************************
   2  * Copyright (c) 2012 Association for Decentralized Information Management
   3  * in Industry THTH ry.
   4  * All rights reserved. This program and the accompanying materials
   5  * are made available under the terms of the Eclipse Public License v1.0
   6  * which accompanies this distribution, and is available at
   7  * http://www.eclipse.org/legal/epl-v10.html
   8  *
   9  * Contributors:
  10  *     VTT Technical Research Centre of Finland - initial API and implementation
  11  *******************************************************************************/
  12 package org.simantics.maps.pojo;
  13
  14 import java.awt.Image;
  15 import java.util.LinkedList;
  16 import java.util.List;
  17 import java.util.concurrent.Executors;
  18 import java.util.concurrent.ScheduledExecutorService;
  19 import java.util.concurrent.TimeUnit;
  20 import java.util.concurrent.atomic.AtomicInteger;
  21
  22 import org.simantics.maps.ProvisionException;
  23 import org.simantics.maps.query.Query;
  24 import org.simantics.maps.tile.IFilter;
  25 import org.simantics.maps.tile.ITileProvider;
  26 import org.simantics.maps.tile.TileKey;
  27 import org.slf4j.Logger;
  28 import org.slf4j.LoggerFactory;
  29
  30 /**
  31  * @author Tuukka Lehtonen
  32  * @see TileJobQueue
  33  */
  34 public class TileJob implements Runnable {
  35
  36     private static final Logger LOGGER = LoggerFactory.getLogger(TileJob.class);
  37     private static final AtomicInteger counter = new AtomicInteger();
  38
  39     private LinkedList<Query<TileKey, Image>> queue = new LinkedList<>();
  40     private ScheduledExecutorService executor = Executors.newScheduledThreadPool(1, runnable -> {
  41         Thread thread = new Thread(runnable, getClass().getSimpleName() + "-" + counter.getAndIncrement());
  42         thread.setDaemon(true);
  43         return thread;
  44     });
  45     private ITileProvider provider;
  46
  47     public TileJob() {
  48     }
  49
  50     public void setTileProvider(ITileProvider provider) {
  51         this.provider = provider;
  52     }
  53
  54     protected Image doQuery(TileKey key) throws ProvisionException {
  55         return provider.get(key);
  56     }
  57
  58     public void run() {
  59         Query<TileKey, Image> job = pullNextJob();
  60         do {
  61             if (job != null) {
  62                 try {
  63                     Image result = doQuery(job.source);
  64                     job.listener.queryComplete(job, result);
  65                 } catch (Exception e) {
  66                     LOGGER.error("Querying failed for job {}", job, e);
  67                     job.listener.queryFailed(job, e);
  68                 }
  69             }
  70             // 4. Pick next job
  71             job = pullNextJob();
  72         } while (job != null);
  73     }
  74
  75     protected synchronized int jobsLeft() {
  76         return queue.size();
  77     }
  78
  79     protected synchronized Query<TileKey, Image> pullNextJob() {
  80         if (queue.isEmpty())
  81             return null;
  82         return queue.removeFirst();
  83     }
  84
  85     public synchronized void clear() {
  86         @SuppressWarnings("unchecked")
  87         Query<TileKey, Image> jobs[] = queue.toArray(new Query[queue.size()]);
  88         for (Query<TileKey, Image> j : jobs)
  89             removeJob(j);
  90     }
  91
  92     public synchronized void addJob(Query<TileKey, Image> job) {
  93         queue.addLast(job);
  94         if (queue.size() == 1) {
  95             executor.execute(this);
  96         }
  97     }
  98
  99     public synchronized void addAsFirstJob(Query<TileKey, Image> job) {
 100         queue.addFirst(job);
 101         if (queue.size() == 1) {
 102             executor.execute(this);
 103         }
 104     }
 105
 106     public synchronized boolean removeJob(Query<TileKey, Image> job) {
 107         if (queue.remove(job)) {
 108             job.listener.queryCanceled(job);
 109             return true;
 110         }
 111         return false;
 112     }
 113
 114     public synchronized void filterQueries(IFilter<TileKey> filter) {
 115         if (queue.isEmpty())
 116             return;
 117
 118         LinkedList<Query<TileKey, Image>> result = new LinkedList<Query<TileKey, Image>>();
 119         for (Query<TileKey, Image> query : queue) {
 120             if (filter.select(query.source))
 121                 result.add(query);
 122             else
 123                 query.listener.queryCanceled(query);
 124         }
 125         this.queue = result;
 126     }
 127
 128     public void dispose() {
 129         executor.shutdown();
 130         try {
 131             if (!executor.awaitTermination(100, TimeUnit.MILLISECONDS)) {
 132                 // this should shutdown for good
 133                 List<Runnable> unfinished = executor.shutdownNow();
 134                 LOGGER.warn("TileJob did not terminate in time - left jobs {}", unfinished);
 135             }
 136         } catch (InterruptedException e) {
 137         }
 138     }
 139
 140 }