bundles/org.simantics.databoard/src/org/simantics/databoard/serialization/ConcurrentSerializerFactory.java

   1 package org.simantics.databoard.serialization;
   2
   3 import java.util.concurrent.CompletableFuture;
   4 import java.util.concurrent.ConcurrentHashMap;
   5 import java.util.concurrent.atomic.AtomicBoolean;
   6
   7 import org.simantics.databoard.binding.Binding;
   8 import org.slf4j.Logger;
   9 import org.slf4j.LoggerFactory;
  10
  11 /**
  12  * @author Jani Simomaa
  13  */
  14 public abstract class ConcurrentSerializerFactory implements SerializerScheme {
  15
  16     private Logger LOGGER = LoggerFactory.getLogger(getClass());
  17
  18     /**
  19      * Repository where serializers are placed.
  20      */
  21     private ConcurrentHashMap<Binding, CompletableFuture<Serializer>> repository = new ConcurrentHashMap<>();
  22     private ConcurrentHashMap<Binding, Serializer> inProgress = new ConcurrentHashMap<>();
  23
  24     /**
  25      * Constructs a serilizer for a binding. Implement this.
  26      * It should use the inprogress -map for construction of
  27      * serializers that have component types.
  28      *
  29      *  e.g.
  30      *   inprogress.put(binding, notCompletelyConstructedSerializer);
  31      *   Serializer componentSerializer = construct( componentBinding );
  32      *   notCompletelyConstructedSerializer.setComponent( componentSerializer );
  33      *   inprogress.remove(binding);
  34      *
  35      * try-finally is not needed.
  36      *
  37      * @param request
  38      * @return
  39      * @throws SerializerConstructionException
  40      */
  41     protected abstract Serializer doConstruct(Binding request) throws SerializerConstructionException;
  42
  43     protected Serializer construct(Binding request) throws SerializerConstructionException {
  44         if (request.cachedSerializer() != null) {
  45             return request.cachedSerializer();
  46         } else {
  47             // this is required cause construct can be called from subclasses
  48             repository.putIfAbsent(request, new CompletableFuture<>());
  49             // lets see if we are processing this already
  50             Serializer binding = inProgress.get(request);
  51             if (binding == null) {
  52                 binding = doConstruct(request);
  53                 request.cacheSerializer(binding);
  54                 CompletableFuture<Serializer> completableFuture = repository.get(request);
  55                 completableFuture.complete(binding); // this releases the waiters
  56             }
  57             return binding;
  58         }
  59     }
  60
  61     protected void addInProgress(Binding request, Serializer serializer) {
  62         inProgress.put(request, serializer);
  63     }
  64
  65     protected void finishInProgress(Binding request) {
  66         Serializer remove = inProgress.remove(request);
  67         request.cacheSerializer(remove);
  68         CompletableFuture<Serializer> existing = repository.get(request);
  69         if (existing != null) {
  70             existing.complete(remove);
  71         } else {
  72             LOGGER.warn("Finishing binding request {} for serializer {} without CompletableFuture!", request, remove);
  73         }
  74     }
  75
  76     @Override
  77     public Serializer getSerializer(Binding binding) throws SerializerConstructionException {
  78         if (binding.cachedSerializer() != null)
  79             return binding.cachedSerializer();
  80         // this shouldn't be called that many time as serializers are cached on bindings
  81         AtomicBoolean shouldConstruct = new AtomicBoolean(false);
  82         CompletableFuture<Serializer> completableFuture = repository.computeIfAbsent(binding, t -> {
  83             shouldConstruct.set(true);
  84             return new CompletableFuture<>();
  85         });
  86         if (shouldConstruct.get()) {
  87             // construct should complete the completable future and release the waiters
  88             construct(binding);
  89         }
  90         // we should be ready at this point
  91         try {
  92             return completableFuture.get();
  93         } catch (Exception e) {
  94             throw new SerializerConstructionException(e);
  95         }
  96     }
  97
  98     public Serializer getSerializerUnchecked(Binding binding) throws RuntimeSerializerConstructionException {
  99         try {
 100             return getSerializer(binding);
 101         } catch (SerializerConstructionException e) {
 102             throw new RuntimeSerializerConstructionException(e);
 103         }
 104     }
 105
 106 }