persolijn

BlobSpliterator.java (5511B)

---

 package osm.protobuf;
 
 import java.io.IOException;
 import java.io.InputStream;
 import java.io.RandomAccessFile;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Spliterator;
 import java.util.concurrent.locks.Lock;
 import java.util.concurrent.locks.ReentrantLock;
 import java.util.function.Consumer;
 import java.util.stream.Stream;
 import java.util.stream.StreamSupport;
 
 import osm.common.RandomAccessFileInputStream;
 import osm.message.Blob;
 import osm.message.BlobHeader;
 import osm.message.Entity;
 import osm.message.HeaderBlock;
 
 /**
  * A {@link Spliterator} implementation for iterating over lists of
  * {@link Entity} objects parsed from OpenStreetMap blobs.
  */
 public class BlobSpliterator implements Spliterator<List<Entity>> {
     /**
      * BlobHeader is never bigger than 64K.
      */
     private static final int MAX_HEADER_SIZE = 64 * 1024;
 
     /**
      * Blob is never bigger than 32M.
      */
     private static final int MAX_BLOB_SIZE = 32 * 1024 * 1024;
 
     private final RandomAccessFile input;
     private final InputStream stream;
     private final Lock lock;
     private final Consumer<HeaderBlock> onHeader;
     private BlobHeader[] headers;
 
     private final int start;
     private int end = 0;
     private int current = 0;
 
     /**
      * Constructs a BlobSpliterator with the specified parameters.
      *
      * @param input    The {@link RandomAccessFile} used for reading blobs.
      * @param onHeader A {@link Consumer} to handle the parsed {@link HeaderBlock}
      *                 objects.
      */
     public BlobSpliterator(RandomAccessFile input, Consumer<HeaderBlock> onHeader) {
         this.input = input;
         this.onHeader = onHeader;
         this.stream = new RandomAccessFileInputStream(input);
         this.lock = new ReentrantLock();
 
         List<BlobHeader> headerList = new ArrayList<>();
         try {
             while (input.getFilePointer() < input.length()) {
                 int headerLength = input.readInt();
 
                 if (headerLength > MAX_HEADER_SIZE)
                     throw new RuntimeException(
                             "blob header exceeds " + MAX_HEADER_SIZE + " bytes (" + headerLength + ")");
 
                 BlobHeader header = new BlobHeader().parse(stream, headerLength);
 
                 if (header.size > MAX_BLOB_SIZE)
                     throw new RuntimeException("blob exceeds " + MAX_BLOB_SIZE + " bytes (" + header.size + ")");
 
                 header.offset = input.getFilePointer();
                 headerList.add(header);
 
                 input.skipBytes(header.size);
             }
         } catch (IOException e) {
             throw new RuntimeException(e);
         }
 
         headers = new BlobHeader[headerList.size()];
         headerList.toArray(headers);
 
         start = 0;
         current = 0;
         end = headers.length;
     }
 
     private BlobSpliterator(RandomAccessFile input, Lock lock, Consumer<HeaderBlock> onHeader, BlobHeader[] headers,
             int start, int end) {
         this.input = input;
         this.lock = lock;
         this.onHeader = onHeader;
         this.stream = new RandomAccessFileInputStream(input);
         this.headers = headers;
         this.start = start;
         this.current = start;
         this.end = end;
     }
 
     @Override
     public Spliterator<List<Entity>> trySplit() {
         // locking because of end
         lock.lock();
         if (current < end - 1)
             return null;
 
         int mid = (end - start) / 2;
         int otherEnd = end;
 
         end = start + mid;
         lock.unlock();
 
         return new BlobSpliterator(input, lock, onHeader, headers, end, otherEnd);
     }
 
     @Override
     public boolean tryAdvance(Consumer<? super List<Entity>> action) {
         // locking because of input, end
         lock.lock();
 
         if (current >= end)
             return false;
 
         BlobHeader header = headers[current++];
 
         try {
             input.seek(header.offset);
         } catch (IOException e) {
             throw new RuntimeException(e);
         }
 
         Blob blob = new Blob(header.headerType).parse(stream, header.size);
 
         // unlocking, every critical variable is unused
         lock.unlock();
 
         if (blob.header != null)
             onHeader.accept(blob.header);
 
         if (blob.primitive != null)
             action.accept(blob.primitive);
 
         return true;
     }
 
     @Override
     public long estimateSize() {
         return end - current;
     }
 
     @Override
     public int characteristics() {
         return DISTINCT | SUBSIZED | SIZED | NONNULL | IMMUTABLE | CONCURRENT;
     }
 
     /**
      * Creates a sequential or parallel {@link Stream} from this
      * {@link BlobSpliterator}.
      *
      * @param parallel A boolean indicating whether the stream should be parallel.
      * @return A {@link Stream} of lists of {@link Entity} objects.
      */
     public Stream<List<Entity>> stream(boolean parallel) {
         return StreamSupport.stream(this, parallel);
     }
 
     /**
      * Creates a sequential stream from this {@link BlobSpliterator}.
      *
      * @return A sequential {@link Stream} of lists of {@link Entity} objects.
      */
     public Stream<List<Entity>> stream() {
         return stream(false);
     }
 
     /**
      * Creates a parallel stream from this {@link BlobSpliterator}.
      *
      * @return A parallel {@link Stream} of lists of {@link Entity} objects.
      */
     public Stream<List<Entity>> parallelStream() {
         return stream(true);
     }
 }