bundles/org.simantics.graph.db/src/org/simantics/graph/db/TransferableGraphs.java

   1 /*******************************************************************************\r
   2  * Copyright (c) 2007, 2010 Association for Decentralized Information Management\r
   3  * in Industry THTH ry.\r
   4  * All rights reserved. This program and the accompanying materials\r
   5  * are made available under the terms of the Eclipse Public License v1.0\r
   6  * which accompanies this distribution, and is available at\r
   7  * http://www.eclipse.org/legal/epl-v10.html\r
   8  *\r
   9  * Contributors:\r
  10  *     VTT Technical Research Centre of Finland - initial API and implementation\r
  11  *******************************************************************************/\r
  12 package org.simantics.graph.db;\r
  13 \r
  14 import java.io.DataInput;\r
  15 import java.io.DataOutput;\r
  16 import java.io.File;\r
  17 import java.io.IOException;\r
  18 import java.io.InputStream;\r
  19 import java.util.ArrayList;\r
  20 import java.util.Collection;\r
  21 import java.util.HashSet;\r
  22 import java.util.TreeMap;\r
  23 import java.util.UUID;\r
  24 \r
  25 import org.simantics.databoard.Accessors;\r
  26 import org.simantics.databoard.Bindings;\r
  27 import org.simantics.databoard.Datatypes;\r
  28 import org.simantics.databoard.accessor.error.AccessorConstructionException;\r
  29 import org.simantics.databoard.accessor.error.AccessorException;\r
  30 import org.simantics.databoard.accessor.file.FileVariantAccessor;\r
  31 import org.simantics.databoard.binding.Binding;\r
  32 import org.simantics.databoard.binding.error.BindingConstructionException;\r
  33 import org.simantics.databoard.binding.error.DatatypeConstructionException;\r
  34 import org.simantics.databoard.binding.mutable.Variant;\r
  35 import org.simantics.databoard.container.DataContainer;\r
  36 import org.simantics.databoard.container.DataContainers;\r
  37 import org.simantics.databoard.serialization.SerializationException;\r
  38 import org.simantics.databoard.serialization.Serializer;\r
  39 import org.simantics.databoard.type.Datatype;\r
  40 import org.simantics.databoard.util.binary.BinaryFile;\r
  41 import org.simantics.databoard.util.binary.RandomAccessBinary;\r
  42 import org.simantics.db.ReadGraph;\r
  43 import org.simantics.db.RequestProcessor;\r
  44 import org.simantics.db.Resource;\r
  45 import org.simantics.db.Session;\r
  46 import org.simantics.db.VirtualGraph;\r
  47 import org.simantics.db.WriteGraph;\r
  48 import org.simantics.db.WriteOnlyGraph;\r
  49 import org.simantics.db.common.CommentMetadata;\r
  50 import org.simantics.db.common.request.ReadRequest;\r
  51 import org.simantics.db.common.request.WriteOnlyRequest;\r
  52 import org.simantics.db.common.request.WriteRequest;\r
  53 import org.simantics.db.exception.CancelTransactionException;\r
  54 import org.simantics.db.exception.DatabaseException;\r
  55 import org.simantics.db.service.SerialisationSupport;\r
  56 import org.simantics.db.service.VirtualGraphSupport;\r
  57 import org.simantics.graph.db.TransferableGraphSource.TransferableGraphSourceValueProcedure;\r
  58 import org.simantics.graph.diff.TransferableGraphDelta1;\r
  59 import org.simantics.graph.representation.Extensions;\r
  60 import org.simantics.graph.representation.External;\r
  61 import org.simantics.graph.representation.Identity;\r
  62 import org.simantics.graph.representation.TransferableGraph1;\r
  63 import org.simantics.graph.representation.Value;\r
  64 import org.simantics.utils.datastructures.BinaryFunction;\r
  65 \r
  66 import gnu.trove.list.array.TIntArrayList;\r
  67 import gnu.trove.map.hash.TObjectIntHashMap;\r
  68 \r
  69 public class TransferableGraphs {\r
  70 \r
  71         public static long[] importGraph(Session session, Object tg, IImportAdvisor advisor) throws DatabaseException, TransferableGraphException {\r
  72                 if (tg instanceof TransferableGraph1) \r
  73                 {\r
  74                         return importGraph1(session, (TransferableGraph1) tg, advisor);\r
  75                 }\r
  76                 throw new TransferableGraphException("Cannot import "+tg.getClass().getName());\r
  77         }\r
  78 \r
  79         public static long[] importGraph(WriteGraph g, Object tg, IImportAdvisor advisor) throws DatabaseException, TransferableGraphException {\r
  80                 if (tg instanceof TransferableGraph1) \r
  81                 {\r
  82                         return importGraph1(g, (TransferableGraph1) tg, advisor);\r
  83                 }\r
  84                 throw new TransferableGraphException("Cannot import "+tg.getClass().getName());\r
  85         }\r
  86         \r
  87         public static long[] importGraph(Session session, Object tg) throws DatabaseException, TransferableGraphException {\r
  88                 if (tg instanceof TransferableGraph1) \r
  89                 {\r
  90                         return importGraph1(session, (TransferableGraph1) tg);\r
  91                 }\r
  92                 throw new TransferableGraphException("Cannot import "+tg.getClass().getName());\r
  93         }\r
  94 \r
  95         public static long[] importGraph(WriteGraph g, Object tg) throws DatabaseException, TransferableGraphException {\r
  96                 if (tg instanceof TransferableGraph1) \r
  97                 {\r
  98                         return importGraph1(g, (TransferableGraph1) tg);\r
  99                 }\r
 100                 throw new TransferableGraphException("Cannot import "+tg.getClass().getName());\r
 101         }\r
 102         \r
 103         public static Collection<Resource> collectExternals(RequestProcessor processor, final TransferableGraph1 tg) throws DatabaseException, TransferableGraphException {\r
 104                 final TransferableGraphImportProcess process = new TransferableGraphImportProcess(tg, new ImportAdvisor());\r
 105                 processor.syncRequest(new ReadRequest() {\r
 106                         @Override\r
 107                         public void run(ReadGraph graph) throws DatabaseException {\r
 108                                 process.prepare(graph);\r
 109                         }\r
 110                 });\r
 111                 HashSet<Resource> result = new HashSet<Resource>();\r
 112                 for(Identity id : tg.identities) {\r
 113                         if(id.definition instanceof External) {\r
 114                                 result.add(process.resources[id.resource]);\r
 115                         }\r
 116                 }\r
 117                 return result;\r
 118         }\r
 119         \r
 120         /**\r
 121          * Imports transferable graph version 1 to the database. Root advisor is used\r
 122          * to give identities to roots of the transferable graphs. It may be null,\r
 123          * in which case new resources are created for all roots but the root library.\r
 124          * \r
 125          * @param session\r
 126          * @param tg\r
 127      * @param advisor root advisor or <code>null</code>\r
 128          * @throws DatabaseException\r
 129          */\r
 130         public static long[] importGraph1(Session session, final TransferableGraph1 tg, final IImportAdvisor advisor_) throws DatabaseException, TransferableGraphException {\r
 131                 \r
 132                 final IImportAdvisor2 advisor = (advisor_ instanceof IImportAdvisor2) ? ((IImportAdvisor2)advisor_) : new WrapperAdvisor(advisor_);\r
 133                 \r
 134                 final TransferableGraphImportProcess process = new TransferableGraphImportProcess(tg, \r
 135                                 advisor == null ? new ImportAdvisor() : advisor);\r
 136                 session.syncRequest(new ReadRequest() {\r
 137                         @Override\r
 138                         public void run(ReadGraph graph) throws DatabaseException {\r
 139                                 process.prepare(graph);\r
 140                         }\r
 141                 });             \r
 142                 session.syncRequest(new WriteOnlyRequest() {\r
 143                         @Override\r
 144                         public void perform(WriteOnlyGraph graph) throws DatabaseException {\r
 145                                 advisor.beforeWrite(graph, process);\r
 146                                 process.write(graph);\r
 147                                 advisor.afterWrite(graph, process);\r
 148                         }\r
 149                 });\r
 150                 return process.getResourceIds(\r
 151                                 session.getService(SerialisationSupport.class));\r
 152         }\r
 153 \r
 154         public static void importGraph1(Session session, final TransferableGraph1 tg, IImportAdvisor advisor, final BinaryFunction<Boolean, WriteOnlyGraph, TransferableGraphImportProcess> callback) throws DatabaseException, TransferableGraphException {\r
 155                 final TransferableGraphImportProcess process = new TransferableGraphImportProcess(tg, \r
 156                                 advisor == null ? new ImportAdvisor() : advisor);\r
 157                 session.syncRequest(new ReadRequest() {\r
 158                         @Override\r
 159                         public void run(ReadGraph graph) throws DatabaseException {\r
 160                                 process.prepare(graph);\r
 161                         }\r
 162                 });\r
 163                 session.syncRequest(new WriteOnlyRequest() {\r
 164                         @Override\r
 165                         public void perform(WriteOnlyGraph graph) throws DatabaseException {\r
 166                                 process.write(graph);\r
 167                                 if(callback != null)\r
 168                                         callback.call(graph, process);\r
 169                         }\r
 170                 });\r
 171         }\r
 172 \r
 173     public static void importGraph1(Session session, final TransferableGraphSource tg, IImportAdvisor advisor) throws Exception {\r
 174         importGraph1(session, tg, advisor, null);\r
 175     }\r
 176 \r
 177     public static void importGraph1(Session session, final TransferableGraphSource tg, IImportAdvisor advisor, TGStatusMonitor monitor) throws DatabaseException {\r
 178         importGraph1(session, null, tg, advisor, monitor);\r
 179     }\r
 180 \r
 181     public static void importGraph1(Session session, VirtualGraph vg, final TransferableGraphSource tg, IImportAdvisor advisor_, TGStatusMonitor monitor) throws DatabaseException {\r
 182         \r
 183         final IImportAdvisor2 advisor = (advisor_ instanceof IImportAdvisor2) ? ((IImportAdvisor2)advisor_) : new WrapperAdvisor(advisor_);\r
 184 \r
 185                 final StreamingTransferableGraphImportProcess process = new StreamingTransferableGraphImportProcess(session, vg, tg, advisor);\r
 186                 session.syncRequest(new ReadRequest() {\r
 187                         @Override\r
 188                         public void run(ReadGraph graph) throws DatabaseException {\r
 189                                 try {\r
 190                                         process.prepare(graph);\r
 191                                 } catch (DatabaseException e) {\r
 192                                         throw e;\r
 193                                 } catch (Exception e) {\r
 194                                         throw new DatabaseException(e);\r
 195                                 }\r
 196                         }\r
 197                 });\r
 198                 session.syncRequest(new WriteOnlyRequest(vg) {\r
 199                         @Override\r
 200                         public void perform(WriteOnlyGraph graph) throws DatabaseException {\r
 201                                 try {\r
 202                                     advisor.beforeWrite(graph, process);\r
 203                                         process.write(graph);\r
 204                     advisor.afterWrite(graph, process);\r
 205                                 } catch (Exception e) {\r
 206                                         throw new DatabaseException(e);\r
 207                                 }\r
 208                         }\r
 209                 });\r
 210         }\r
 211 \r
 212         public static void importGraph1WithMonitor(Session session, final TransferableGraph1 tg, IImportAdvisor advisor_, TGStatusMonitor monitor) throws DatabaseException {\r
 213                 final IImportAdvisor2 advisor = (advisor_ instanceof IImportAdvisor2) ? ((IImportAdvisor2)advisor_) : new WrapperAdvisor(advisor_);\r
 214                 final TransferableGraphImportProcess process = new TransferableGraphImportProcess(tg, \r
 215                                 advisor == null ? new ImportAdvisor() : advisor, monitor);\r
 216                 session.syncRequest(new ReadRequest() {\r
 217                         @Override\r
 218                         public void run(ReadGraph graph) throws DatabaseException {\r
 219                                 process.prepare(graph);\r
 220                         }\r
 221                 });\r
 222                 session.syncRequest(new WriteOnlyRequest() {\r
 223                         @Override\r
 224                         public void perform(WriteOnlyGraph graph) throws DatabaseException {\r
 225                                 advisor.beforeWrite(graph, process);\r
 226                                 process.write2(graph);\r
 227                                 advisor.afterWrite(graph, process);\r
 228                                 CommentMetadata comments = graph.getMetadata(CommentMetadata.class);\r
 229                                 comments.add("Imported transferable graph with " + tg.resourceCount + " resources");\r
 230                                 graph.addMetadata(comments);\r
 231                         }\r
 232                 });\r
 233         }\r
 234 \r
 235         public static void importGraph1WithChanges(Session session, final TransferableGraph1 tg, IImportAdvisor advisor, final BinaryFunction<Boolean, WriteGraph, TransferableGraphImportProcess> callback) throws DatabaseException, TransferableGraphException {\r
 236                 final TransferableGraphImportProcess process = new TransferableGraphImportProcess(tg, \r
 237                                 advisor == null ? new ImportAdvisor() : advisor);\r
 238                 session.syncRequest(new ReadRequest() {\r
 239                         @Override\r
 240                         public void run(ReadGraph graph) throws DatabaseException {\r
 241                                 process.prepare(graph);\r
 242                         }\r
 243                 });\r
 244                 session.syncRequest(new WriteRequest() {\r
 245                         @Override\r
 246                         public void perform(WriteGraph graph) throws DatabaseException {\r
 247                                 process.write2(graph);\r
 248                 CommentMetadata comments = graph.getMetadata(CommentMetadata.class);\r
 249                 comments.add("Imported transferable graph with " + tg.resourceCount + " resources");\r
 250                 graph.addMetadata(comments);\r
 251                                 if(callback != null)\r
 252                                         callback.call(graph, process);\r
 253                         }\r
 254                 });\r
 255         }\r
 256         \r
 257         public static long[] importGraph1(Session session, final TransferableGraph1 tg) throws DatabaseException, TransferableGraphException {\r
 258                 final TransferableGraphImportProcess process = \r
 259                         new TransferableGraphImportProcess(tg, \r
 260                                 null);\r
 261                 session.syncRequest(new ReadRequest() {\r
 262                         @Override\r
 263                         public void run(ReadGraph graph) throws DatabaseException {\r
 264                                 process.prepare(graph);\r
 265                         }\r
 266                 });             \r
 267                 session.syncRequest(new WriteOnlyRequest() {\r
 268                         @Override\r
 269                         public void perform(WriteOnlyGraph graph) throws DatabaseException {\r
 270                                 process.write(graph);\r
 271                         }\r
 272                 });\r
 273                 return process.getResourceIds(\r
 274                                 session.getService(SerialisationSupport.class));\r
 275         }\r
 276         \r
 277         public static long[] importGraph1(WriteGraph graph, final TransferableGraph1 tg) throws DatabaseException, TransferableGraphException {\r
 278                 final TransferableGraphImportProcess process = \r
 279                         new TransferableGraphImportProcess(tg, \r
 280                                 null);\r
 281                 process.prepare(graph);\r
 282                 process.write2(graph);\r
 283                 return process.getResourceIds(\r
 284                                 graph.getSession().getService(SerialisationSupport.class));\r
 285         }\r
 286         \r
 287         /**\r
 288          * Import transferable graph version 1 to the database. Root advisor is used\r
 289          * to give identities to roots of the transferable graphs. It may be null,\r
 290          * in which case new resources are created for all roots but the root library.\r
 291          * \r
 292          * @param session\r
 293          * @param tg\r
 294      * @param advisor root advisor or <code>null</code>\r
 295          * @throws DatabaseException\r
 296          */\r
 297         public static long[] importGraph1(WriteGraph graph, TransferableGraph1 tg, IImportAdvisor advisor) throws DatabaseException {\r
 298             return importGraph1(graph, tg, advisor, null);\r
 299         }\r
 300 \r
 301         public static long[] importGraph1(WriteGraph graph, TransferableGraph1 tg, IImportAdvisor advisor, TGStatusMonitor monitor) throws DatabaseException {\r
 302                 TransferableGraphImportProcess process = new TransferableGraphImportProcess(tg, \r
 303                                 advisor == null ? new ImportAdvisor() : advisor, monitor);\r
 304                 process.prepare(graph);\r
 305                 if(advisor instanceof IImportAdvisor2) ((IImportAdvisor2)advisor).beforeWrite(graph, process);\r
 306                 process.write2(graph);\r
 307                 if(advisor instanceof IImportAdvisor2) ((IImportAdvisor2)advisor).afterWrite(graph, process);\r
 308                 return process.getResourceIds(\r
 309                                 graph.getSession().getService(SerialisationSupport.class));\r
 310         }\r
 311 \r
 312         public static long[] applyDelta(WriteGraph graph, long[] oldResources, TransferableGraphDelta1 delta) throws DatabaseException {\r
 313                 SerialisationSupport serializer = \r
 314                         graph.getSession().getService(SerialisationSupport.class);\r
 315                 \r
 316                 TGToGraphMap aMap = new TGToGraphMap(delta.a);\r
 317                 aMap.addOldResources(serializer, oldResources);\r
 318                 aMap.deny(graph);\r
 319                 \r
 320                 TGToGraphMap bMap = new TGToGraphMap(delta.b);\r
 321                 bMap.addMappedOldResources(serializer, delta.aToB, aMap.getResources());\r
 322                 bMap.prepare(graph);\r
 323                 bMap.claim(graph);\r
 324                 \r
 325                 return bMap.getResources(serializer);\r
 326         }\r
 327         \r
 328         public static boolean hasChanges(ReadGraph graph, long[] oldResources, TransferableGraphDelta1 delta) throws DatabaseException {\r
 329                 \r
 330                 SerialisationSupport serializer = \r
 331                         graph.getSession().getService(SerialisationSupport.class);\r
 332                 \r
 333                 TGToGraphMap aMap = new TGToGraphMap(delta.a);\r
 334                 aMap.addOldResources(serializer, oldResources);\r
 335                 if(aMap.checkDeny(graph)) return true;\r
 336                 \r
 337                 TGToGraphMap bMap = new TGToGraphMap(delta.b);\r
 338                 bMap.addMappedOldResources(serializer, delta.aToB, aMap.getResources());\r
 339                 bMap.prepare(graph);\r
 340                 return bMap.checkClaim(graph);\r
 341                 \r
 342         }\r
 343 \r
 344         public static void uninstallGraph(WriteGraph writeGraph, TransferableGraph1 graph,\r
 345                         ImportAdvisor advisor) throws TransferableGraphException {\r
 346                 // TODO HANNU IMPLEMENTS\r
 347                 throw new UnsupportedOperationException();\r
 348         }       \r
 349         \r
 350         public static long[] importVirtualGraph(Session session, final VirtualGraph vg, final TransferableGraph1 tg, IImportAdvisor advisor) throws DatabaseException {\r
 351                 final TransferableGraphImportProcess process = new TransferableGraphImportProcess(tg, \r
 352                                 advisor == null ? new ImportAdvisor() : advisor);\r
 353                 session.syncRequest(new ReadRequest() {\r
 354                         @Override\r
 355                         public void run(ReadGraph graph) throws DatabaseException {\r
 356                                 process.prepare(graph);\r
 357                         }\r
 358                 });\r
 359                 session.syncRequest(new WriteOnlyRequest(vg) {\r
 360                         @Override\r
 361                         public void perform(WriteOnlyGraph graph) throws DatabaseException {\r
 362                                 // Needed because process#write does not support virtual WriteOnlyGraph\r
 363                                 if (vg != null)\r
 364                                         process.write2(graph);\r
 365                                 else\r
 366                                         process.write(graph);\r
 367                         }\r
 368                         \r
 369                 });\r
 370                 return process.getResourceIds(session.getService(SerialisationSupport.class));\r
 371         }       \r
 372 \r
 373         public static long[] importVirtualGraph(WriteGraph graph, final TransferableGraph1 tg, IImportAdvisor advisor) throws DatabaseException {\r
 374                 final TransferableGraphImportProcess process = new TransferableGraphImportProcess(tg, \r
 375                                 advisor == null ? new ImportAdvisor() : advisor);\r
 376                 process.prepare(graph);\r
 377                 process.write2(graph);\r
 378                 return process.getResourceIds(graph.getService(SerialisationSupport.class));\r
 379         }       \r
 380         \r
 381         public static TransferableGraph1 readGraph(File file) throws TransferableGraphException {\r
 382                 FileVariantAccessor va = null;\r
 383                 try {\r
 384                         va = Accessors.openAccessor(file);\r
 385                         Datatype type = va.getContentType();\r
 386                         if(type.equals(Datatypes.getDatatype(TransferableGraph1.class))) \r
 387                                 return  (TransferableGraph1)va.getContentValue(Bindings.getBinding(TransferableGraph1.class));\r
 388                         else\r
 389                                 throw new SerializationException("Unknown transferable graph data type.");\r
 390                 } catch (AccessorException e) {\r
 391                         throw new TransferableGraphException(e);\r
 392                 } catch (BindingConstructionException e) {\r
 393                         throw new TransferableGraphException(e);\r
 394                 } catch (SerializationException e) {\r
 395                         throw new TransferableGraphException(e);\r
 396                 } catch (AccessorConstructionException e) {\r
 397                         throw new TransferableGraphException(e);\r
 398                 } catch (DatatypeConstructionException e) {\r
 399                         throw new TransferableGraphException(e);\r
 400                 } finally {\r
 401                         if(va != null) {\r
 402                                 try {\r
 403                                         va.close();\r
 404                                 } catch (AccessorException e) {\r
 405                                 }\r
 406                         }\r
 407                 } \r
 408         }\r
 409         \r
 410         public static void importVirtualGraph(Session session, VirtualGraph vg, File file) throws TransferableGraphException {\r
 411                 try {\r
 412                         importVirtualGraph(session, vg, readGraph(file), new ImportAdvisor());\r
 413                 } catch (DatabaseException e) {\r
 414                         throw new TransferableGraphException(e);\r
 415                 }\r
 416         }\r
 417         \r
 418         public static VirtualGraph importVirtualGraph(Session session, File file) throws TransferableGraphException {\r
 419                 VirtualGraphSupport support = session.getService(VirtualGraphSupport.class);\r
 420                 VirtualGraph vg = support.getMemoryPersistent(UUID.randomUUID().toString());\r
 421                 importVirtualGraph(session, vg, file);\r
 422                 return vg;\r
 423         }\r
 424         \r
 425         public static void writeTransferableGraph(RequestProcessor processor, final String format, final TransferableGraphSource source, File target) throws Exception {\r
 426                 writeTransferableGraph(processor, format, 1, source, target);\r
 427         }\r
 428 \r
 429         public static void writeTransferableGraph(RequestProcessor processor, final String format, final int version, final TransferableGraphSource source, File target) throws Exception {\r
 430                 writeTransferableGraph(processor, format, version, new TreeMap<String,Variant>(), source, target);\r
 431         }\r
 432 \r
 433         public static void writeTransferableGraph(RequestProcessor processor, String format, int version, TreeMap<String, Variant> metadata, TransferableGraphSource source, File target) throws Exception {\r
 434                 writeTransferableGraph(processor, format, version, metadata, source, target, TGStatusMonitor.NULL_MONITOR);\r
 435         }\r
 436 \r
 437         public static void writeTransferableGraph(RequestProcessor processor, String format, int version, TreeMap<String, Variant> metadata, TransferableGraphSource source, File target, TGStatusMonitor monitor) throws Exception {\r
 438                 final Serializer datatypeSerializer = Bindings.getSerializerUnchecked(Datatype.class);\r
 439                 try (RandomAccessBinary out = new BinaryFile(target, 128*1024)) {\r
 440                         DataContainer container = new DataContainer(format, version, metadata, null);\r
 441                         DataContainers.writeHeader(out, container);\r
 442                         datatypeSerializer.serialize((DataOutput) out, Datatypes.getDatatypeUnchecked(TransferableGraph1.class));\r
 443                         writeTransferableGraph(processor, source, out, monitor);\r
 444                 }\r
 445         }\r
 446 \r
 447         private static TGStatusMonitor safeMonitor(TGStatusMonitor mon) {\r
 448                 return mon == null ? TGStatusMonitor.NULL_MONITOR : mon;\r
 449         }\r
 450 \r
 451         private static class CopyingInputStream extends InputStream {\r
 452                 public DataInput in;\r
 453                 public DataOutput out;\r
 454 \r
 455                 @Override\r
 456                 public int read() throws IOException {\r
 457                         int value = in.readUnsignedByte();\r
 458                         out.write(value);\r
 459                         return value;\r
 460                 }\r
 461         }\r
 462 \r
 463         private static long copy(byte[] buffer, DataInput in, DataOutput out, long bytesToCopy) throws IOException {\r
 464                 int read = 0;\r
 465                 long bufferLength = buffer.length;\r
 466                 while (read < bytesToCopy) {\r
 467                         int l = (int) Math.min(bufferLength, bytesToCopy-read);\r
 468                         in.readFully(buffer, 0, l);\r
 469                         out.write(buffer, 0, l);\r
 470                         read += l;\r
 471                 }\r
 472                 return read;\r
 473         }\r
 474 \r
 475         private static final int LITERAL_VALUE_IO_BUFFER_SIZE = 128 * 1024;\r
 476 \r
 477         private static void writeTransferableGraph(RequestProcessor processor, final TransferableGraphSource source, final RandomAccessBinary out, TGStatusMonitor monitor) throws Exception {\r
 478                 long start = System.nanoTime();\r
 479 \r
 480                 final Serializer datatypeSerializer = Bindings.getSerializerUnchecked(Datatype.class);\r
 481                 final Serializer identitySerializer = Bindings.getSerializerUnchecked(Identity.class);\r
 482                 final Serializer extensionSerializer = Bindings.getSerializerUnchecked(Extensions.class);\r
 483 \r
 484                 int resourceCount = source.getResourceCount();\r
 485                 //System.err.println("resourceCount: " + resourceCount);\r
 486                 out.writeInt(resourceCount);\r
 487                 extensionSerializer.serialize(out, new Extensions(source.getExtensions()));\r
 488 \r
 489 //              System.err.println("resource count: " + source.getResourceCount());\r
 490 //              System.err.println("identity count: " + source.getIdentityCount());\r
 491 \r
 492                 byte[] buffer = new byte[LITERAL_VALUE_IO_BUFFER_SIZE];\r
 493 \r
 494                 processor.syncRequest(new ReadRequest() {\r
 495                         @Override\r
 496                         public void run(ReadGraph graph) throws DatabaseException {\r
 497                                 try {\r
 498                                         if (monitor.isCanceled())\r
 499                                                 throw new CancelTransactionException();\r
 500 \r
 501                                         int identityCount = source.getIdentityCount();\r
 502                                         TGStatusMonitor.Updater identityProgress = new TGStatusMonitor.Updater(safeMonitor(monitor), 0, 33, identityCount);\r
 503                                         out.writeInt(identityCount);\r
 504                                         //System.err.println("identities: " + identityCount);\r
 505                                         source.forIdentities(graph, value -> {\r
 506                                                 //System.err.println("id: " + value);\r
 507                                                 identitySerializer.serialize(out, value);\r
 508                                                 identityProgress.worked(1);\r
 509                                         });\r
 510 \r
 511                                         if (monitor.isCanceled())\r
 512                                                 throw new CancelTransactionException();\r
 513 \r
 514                                         long statementCountPos = out.position();\r
 515                                         int originalStatementCount = source.getStatementCount();\r
 516                                         TGStatusMonitor.Updater statementProgress = new TGStatusMonitor.Updater(safeMonitor(monitor), 34, 66, originalStatementCount);\r
 517                                         out.writeInt(originalStatementCount);\r
 518                                         //System.err.println("original statementCount: " + originalStatementCount);\r
 519                                         int[] statementCounter = { 0 };\r
 520                                         source.forStatements(graph, r -> {\r
 521                                                 for (int i = 0; i < 4; ++i)\r
 522                                                         out.writeInt(r[i]);\r
 523                                                 statementCounter[0]++;\r
 524                                                 //System.err.println("stm " + (statementCounter[0]) + ": " + r[0] + " " + r[1] + " " + r[2] + " " + r[3]);\r
 525                                                 statementProgress.worked(1);\r
 526                                         });\r
 527                                         //System.err.println("wrote " + statementCounter[0] + " statements, " + (statementCounter[0]*4)+ " integers");\r
 528 \r
 529                                         // Rewrite statement count after knowing exactly how many\r
 530                                         // statements were written. It is possible that some\r
 531                                         // statements get filtered out at this stage and the\r
 532                                         // original statement count does not reflect that.\r
 533                                         long afterStatementsPos = out.position();\r
 534                                         out.position(statementCountPos);\r
 535                                         out.writeInt(statementCounter[0]*4);\r
 536                                         out.position(afterStatementsPos);\r
 537 \r
 538                                         if (monitor.isCanceled())\r
 539                                                 throw new CancelTransactionException();\r
 540 \r
 541                                         int valueCount = source.getValueCount();\r
 542                                         TGStatusMonitor.Updater valueProgress = new TGStatusMonitor.Updater(safeMonitor(monitor), 67, 100, valueCount);\r
 543                                         out.writeInt(valueCount);\r
 544 //                                      System.err.println("valueCount: " + valueCount);\r
 545                                         CopyingInputStream cis = new CopyingInputStream();\r
 546                                         cis.out = out;\r
 547                                         source.forValues2(graph, new TransferableGraphSourceValueProcedure() {\r
 548                                                 TObjectIntHashMap<Object> identities = new TObjectIntHashMap<>();\r
 549 \r
 550                                                 @Override\r
 551                                                 public void rawCopy(int resource, int length, DataInput input) throws Exception {\r
 552                                                         out.writeInt(resource);\r
 553                                                         long copied = copy(buffer, input, out, length);\r
 554                                                         assert copied == length;\r
 555                                                         //System.err.println("value " + (num++) + ": raw variant, " + length + " bytes, copied " + copied + " bytes");\r
 556                                                         valueProgress.worked(1);\r
 557                                                 }\r
 558 \r
 559                                                 @Override\r
 560                                                 public void execute(int resource, Datatype type, DataInput input) throws Exception {\r
 561                                                         out.writeInt(resource);\r
 562                                                         identities.clear();\r
 563                                                         datatypeSerializer.serialize(out, identities, type);\r
 564                                                         Binding binding = Bindings.getBinding(type);\r
 565                                                         Serializer serializer = Bindings.getSerializer(binding);\r
 566                                                         cis.in = input;\r
 567                                                         serializer.skip(cis);\r
 568                                                         cis.in = null;\r
 569                                                         valueProgress.worked(1);\r
 570                                                 }\r
 571                                         });\r
 572 \r
 573                                 } catch (DatabaseException e) {\r
 574                                         throw e;\r
 575                                 } catch (Exception e) {\r
 576                                         throw new DatabaseException(e);\r
 577                                 }\r
 578                         }\r
 579                 });\r
 580 \r
 581                 long end = System.nanoTime();\r
 582                 System.err.println("Wrote transferable graph in " + 1e-9*(end-start) + " seconds.");\r
 583         }\r
 584 \r
 585         public static TransferableGraph1 create(ReadGraph graph, TransferableGraphSource source) throws DatabaseException {\r
 586                 \r
 587                 final TIntArrayList statements = new TIntArrayList();\r
 588                 final ArrayList<Value> values = new ArrayList<>();\r
 589                 final ArrayList<Identity> identities = new ArrayList<>();\r
 590                 \r
 591                 try {\r
 592 \r
 593                         source.forStatements(graph, r -> statements.addAll(r));\r
 594                         source.forValues(graph, v -> values.add(v));\r
 595                         source.forIdentities(graph, i -> identities.add(i));\r
 596 \r
 597                         return new TransferableGraph1(source.getResourceCount(), \r
 598                                         identities.toArray(new Identity[identities.size()]),\r
 599                                         statements.toArray(),\r
 600                                         values.toArray(new Value[values.size()]),\r
 601                                         source.getExtensions());\r
 602                         \r
 603                 } catch (Exception e) {\r
 604                         \r
 605                         throw new DatabaseException(e);\r
 606                         \r
 607                 }\r
 608                 \r
 609         }\r
 610 \r
 611 }\r