/** We always add one primitive group of less that 8k elements to each primitive block. */
private void endBlock () {
    if (nEntitiesInBlock > 0) {
        if (currEntityType == OSMEntity.Type.NODE) {
            primitiveGroupBuilder.setDense(denseNodesBuilder);
        }
        // Pass the block off to the compression/writing thread
        try {
            Osmformat.PrimitiveBlock primitiveBlock = Osmformat.PrimitiveBlock.newBuilder()
                    .setStringtable(stringTable.toBuilder()).addPrimitivegroup(primitiveGroupBuilder).build();
            synchronousQueue.put(primitiveBlock);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }
}
 

@Override
public void writeBegin() throws IOException {

    LOG.info("Writing PBF format...");

    // Write out a header block
    Osmformat.HeaderBlock.Builder builder = Osmformat.HeaderBlock.newBuilder();
    builder.addRequiredFeatures("DenseNodes").setWritingprogram("Vanilla Extract").build();
    if (timestamp > 0) {
        builder.setOsmosisReplicationTimestamp(timestamp);
    }
    writeOneBlob(builder.build());

    // Start another thread that will handle compression and writing in parallel.
    writerThread = new Thread(this);
    writerThread.start();

}
 

@Override
public void writeWay(long id, Way way) throws IOException {

    checkBlockTransition(OSMEntity.Type.WAY);
    Osmformat.Way.Builder builder = Osmformat.Way.newBuilder().setId(id);

    /* Tags */
    if (way.tags != null) {
        for (OSMEntity.Tag tag : way.tags) {
            if (tag.value == null) tag.value = "";
            builder.addKeys(stringTable.getCode(tag.key));
            builder.addVals(stringTable.getCode(tag.value));
        }
    }

    /* Node References */
    long prevNodeRef = 0;
    for (long ref : way.nodes) {
        builder.addRefs(ref - prevNodeRef); // delta-coded node references
        prevNodeRef = ref;
    }

    /* TODO Should we trigger the build here or just call with the builder? */
    primitiveGroupBuilder.addWays(builder.build());

}
 

/** Runnable interface implementation that compresses and writes output blocks asynchronously. */
@Override
public void run() {
    while (true) {
        try {
            Osmformat.PrimitiveBlock block = synchronousQueue.take(); // block until work is available
            if (block.getPrimitivegroupCount() == 0) {
                break; // a block with no primitive groups tells the writer thread to shut down.
            }
            writeOneBlob(block);
        } catch (InterruptedException ex) {
            LOG.error("Block writer thread was interrupted while waiting for work.");
            break;
        }
    }
    try {
        downstream.flush();
        downstream.close();
    } catch (IOException e) {
        throw new RuntimeException(e);
    }
}
 

/** Note that in many PBF files this function is never called because all nodes are dense. */
@Override
protected void parseNodes(List<Osmformat.Node> nodes) {
    try {
        for (Osmformat.Node n : nodes) {
            if (nodeCount++ % 10000000 == 0) {
                LOG.info("node {}", human(nodeCount));
            }
            Node node = new Node(parseLat(n.getLat()), parseLon(n.getLon()));
            for (int k = 0; k < n.getKeysCount(); k++) {
                String key = getStringById(n.getKeys(k));
                String val = getStringById(n.getVals(k));
                if (retainTag(key)) node.addTag(key, val);
            }
            entitySink.writeNode(n.getId(), node);
        }
    } catch (IOException ex) {
        LOG.error("An I/O exception occurred in the OSM entity sink.");
        ex.printStackTrace();
    }
}
 

@Override
public void parse(Osmformat.HeaderBlock block) {
    for (String s : block.getRequiredFeaturesList()) {
        if (s.equals("OsmSchema-V0.6")) {
            continue; // We can parse this.
        }
        if (s.equals("DenseNodes")) {
            continue; // We can parse this.
        }
        throw new IllegalStateException("File requires unknown feature: " + s);
    }
    if (block.hasOsmosisReplicationTimestamp()) {
        long timestamp = block.getOsmosisReplicationTimestamp();
        LOG.info("PBF file has a replication timestamp of {}", Instant.ofEpochSecond(timestamp));
        entitySink.setReplicationTimestamp(timestamp);
    } else {
        LOG.info("PBF file has no replication timestamp.");
    }
}
 

private void processWays( List<Osmformat.Way> ways, BPbfFieldDecoder fieldDecoder )
{
    for (Osmformat.Way way : ways)
    {
        Map<String, String> tags = buildTags(way.getKeysList(), way.getValsList(), fieldDecoder);

        // Build up the list of way nodes for the way. The node ids are
        // delta encoded meaning that each id is stored as a delta against
        // the previous one.
        long nodeId = 0;
        LongList wayNodes = new LongList( 16 );
        for (long nodeIdOffset : way.getRefsList())
        {
            nodeId += nodeIdOffset;
            wayNodes.add(nodeId);
        }

        parser.addWay( way.getId(), tags, wayNodes );
    }
}
 

@Override
protected void parseNodes(final List<Osmformat.Node> nodes) {
  for (final Osmformat.Node n : nodes) {
    final AvroNode n2 = new AvroNode();
    final AvroPrimitive p = getPrimitive(n.getInfo());
    p.setId(n.getId());
    p.setTags(getTags(n.getKeysList(), n.getValsList()));
    n2.setCommon(p);
    n2.setLatitude(parseLat(n.getLat()));
    n2.setLongitude(parseLon(n.getLon()));
    try {
      nodeWriter.append(n2);
    } catch (final IOException e) {
      LOGGER.error("Unable to write node", e);
    }
  }
}
 

@Override
protected void parseWays(final List<Osmformat.Way> ways) {
  for (final Osmformat.Way w : ways) {
    final AvroWay w2 = new AvroWay();
    final AvroPrimitive p = getPrimitive(w.getInfo());
    p.setId(w.getId());
    p.setTags(getTags(w.getKeysList(), w.getValsList()));
    w2.setCommon(p);

    long lastRef = 0;
    final List<Long> nodes = new ArrayList<>(w.getRefsCount());
    for (final Long ref : w.getRefsList()) {
      lastRef += ref;
      nodes.add(lastRef);
    }
    w2.setNodes(nodes);

    try {
      wayWriter.append(w2);
    } catch (final IOException e) {
      LOGGER.error("Unable to write way", e);
    }
  }
}
 

private AvroPrimitive getPrimitive(final Osmformat.Info info) {
  final AvroPrimitive p = new AvroPrimitive();
  p.setVersion((long) info.getVersion());
  p.setTimestamp(info.getTimestamp());
  p.setUserId((long) info.getUid());
  try {
    p.setUserName(getStringById(info.getUid()));
  } catch (final Exception ex) {
    LOGGER.warn(
        "Error, input file doesn't contain a valid string table for user id: " + info.getUid(),
        ex);
    p.setUserName(String.valueOf(info.getUid()));
  }
  p.setChangesetId(info.getChangeset());
  p.setVisible(info.getVisible());
  return p;
}
 

/** Reset the inter-entity delta coding values and set up a new block. */
private void beginBlock(OSMEntity.Type eType) throws IOException {
    prevId = prevFixedLat = prevFixedLon = nEntitiesInBlock = 0;
    stringTable.clear();
    primitiveGroupBuilder = Osmformat.PrimitiveGroup.newBuilder();
    if (eType == OSMEntity.Type.NODE) {
        denseNodesBuilder = Osmformat.DenseNodes.newBuilder();
    }
}
 

/** @param block is either a PrimitiveBlock or a HeaderBlock */
private void writeOneBlob(GeneratedMessageLite block) {

    // FIXME lotsa big copies going on here

    String blobTypeString;
    if (block instanceof Osmformat.HeaderBlock) {
        blobTypeString = "OSMHeader";
    } else if (block instanceof Osmformat.PrimitiveBlock) {
        blobTypeString = "OSMData";
    } else {
        throw new AssertionError("block must be either a header block or a primitive block.");
    }

    Fileformat.Blob.Builder blobBuilder = Fileformat.Blob.newBuilder();
    byte[] serializedBlock = block.toByteArray();
    byte[] deflatedBlock = new byte[serializedBlock.length];
    int deflatedSize = deflate(serializedBlock, deflatedBlock);
    if (deflatedSize < 0) {
        LOG.debug("Deflate did not reduce the size of a block. Saving it uncompressed.");
        blobBuilder.setRaw(ByteString.copyFrom(serializedBlock));
    } else {
        blobBuilder.setZlibData(ByteString.copyFrom(deflatedBlock, 0, deflatedSize));
        blobBuilder.setRawSize(serializedBlock.length);
    }
    byte[] serializedBlob = blobBuilder.build().toByteArray();

    Fileformat.BlobHeader blobHeader = Fileformat.BlobHeader.newBuilder()
            .setType(blobTypeString).setDatasize(serializedBlob.length).build();
    byte[] serializedBlobHeader = blobHeader.toByteArray();
    try {
        // "Returns a big-endian representation of value in a 4-element byte array"
        downstream.write(Ints.toByteArray(serializedBlobHeader.length));
        downstream.write(serializedBlobHeader);
        downstream.write(serializedBlob);
    } catch (IOException e) {
        throw new RuntimeException(e);
    }

}
 

@Override
public void writeEnd() throws IOException {
    // Finish any partially-completed block.
    endBlock();
    // Send a primitive block with no primitive group to the writer thread, signaling it to shut down and clean up.
    try {
        synchronousQueue.put(Osmformat.PrimitiveBlock.getDefaultInstance());
        writerThread.join();
    } catch (InterruptedException e) {
        throw new RuntimeException(e);
    }
    LOG.info("Finished writing PBF format.");
}
 

@Override
public void writeRelation(long id, Relation relation) throws IOException {

    checkBlockTransition(OSMEntity.Type.RELATION);
    Osmformat.Relation.Builder builder = Osmformat.Relation.newBuilder().setId(id);

    /* Tags */
    if (relation.tags != null) {
        for (OSMEntity.Tag tag : relation.tags) {
            if (tag.value == null) tag.value = "";
            builder.addKeys(stringTable.getCode(tag.key));
            builder.addVals(stringTable.getCode(tag.value));
        }
    }

    /* Relation members */
    long lastMemberId = 0;
    for (Relation.Member member : relation.members) {
        builder.addMemids(member.id - lastMemberId); // delta-coded member references
        lastMemberId = member.id;
        builder.addRolesSid(stringTable.getCode(member.role));
        Osmformat.Relation.MemberType memberType;
        if (member.type == OSMEntity.Type.NODE) {
            memberType = Osmformat.Relation.MemberType.NODE;
        } else if (member.type == OSMEntity.Type.WAY) {
            memberType = Osmformat.Relation.MemberType.WAY;
        } else if (member.type == OSMEntity.Type.RELATION) {
            memberType = Osmformat.Relation.MemberType.RELATION;
        } else {
            throw new RuntimeException("Member type was not defined.");
        }
        builder.addTypes(memberType);
    }

    /* TODO Should we trigger the build here or just call with the builder? */
    primitiveGroupBuilder.addRelations(builder.build());

}
 

public Osmformat.StringTable.Builder toBuilder () {
    Osmformat.StringTable.Builder builder = Osmformat.StringTable.newBuilder();
    for (String s : stringForCode) {
        builder.addS(ByteString.copyFromUtf8(s));
    }
    return builder;
}
 

/**
 * Nodes are usually stored this way. Dense nodes use parallel arrays (a column store) to defeat typical
 * Protobuf message structure.
 */
@Override
protected void parseDense(Osmformat.DenseNodes nodes) {
    long lastId = 0, lastLat = 0, lastLon = 0;
    int kv = 0; // index into the keysvals array
    try {
        for (int n = 0; n < nodes.getIdCount(); n++) {
            if (nodeCount++ % 5000000 == 0) {
                LOG.info("node {}", human(nodeCount));
            }
            Node node = new Node();
            long id = nodes.getId(n) + lastId;
            long lat = nodes.getLat(n) + lastLat;
            long lon = nodes.getLon(n) + lastLon;
            lastId = id;
            lastLat = lat;
            lastLon = lon;
            node.setLatLon(parseLat(lat), parseLon(lon));
            // Check whether any node has tags.
            if (nodes.getKeysValsCount() > 0) {
                while (nodes.getKeysVals(kv) != 0) {
                    int kid = nodes.getKeysVals(kv++);
                    int vid = nodes.getKeysVals(kv++);
                    String key = getStringById(kid);
                    String val = getStringById(vid);
                    if (retainTag(key)) node.addTag(key, val);
                }
                kv++; // Skip over the '0' delimiter.
            }
            entitySink.writeNode(id, node);
        }
    } catch (IOException ex) {
        LOG.error("An I/O exception occurred in the OSM entity sink.");
        ex.printStackTrace();
    }
}
 

@Override
protected void parseWays(List<Osmformat.Way> ways) {
    try {
        for (Osmformat.Way w : ways) {
            if (wayCount++ % 1000000 == 0) {
                LOG.info("way {}", human(wayCount));
            }
            Way way = new Way();
            /* Handle tags */
            for (int k = 0; k < w.getKeysCount(); k++) {
                String key = getStringById(w.getKeys(k));
                String val = getStringById(w.getVals(k));
                if (retainTag(key)) way.addTag(key, val);
            }
            /* Handle nodes */
            List<Long> rl = w.getRefsList();
            long[] nodes = new long[rl.size()];
            long ref = 0;
            for (int n = 0; n < nodes.length; n++) {
                ref += rl.get(n);
                nodes[n] = ref;
            }
            way.nodes = nodes;
            entitySink.writeWay(w.getId(), way);
        }
    } catch (IOException ex) {
        LOG.error("An I/O exception occurred in the OSM entity sink.");
        ex.printStackTrace();
    }
}
 

private void processOsmHeader( byte[] data ) throws InvalidProtocolBufferException
{
    Osmformat.HeaderBlock header = Osmformat.HeaderBlock.parseFrom(data);

    // Build the list of active and unsupported features in the file.
    List<String> supportedFeatures = Arrays.asList("OsmSchema-V0.6", "DenseNodes");
    List<String> activeFeatures = new ArrayList<String>();
    List<String> unsupportedFeatures = new ArrayList<String>();
    for (String feature : header.getRequiredFeaturesList())
    {
        if (supportedFeatures.contains(feature))
        {
            activeFeatures.add(feature);
        } else
        {
            unsupportedFeatures.add(feature);
        }
    }

    // We can't continue if there are any unsupported features. We wait
    // until now so that we can display all unsupported features instead of
    // just the first one we encounter.
    if (unsupportedFeatures.size() > 0)
    {
        throw new RuntimeException("PBF file contains unsupported features " + unsupportedFeatures);
    }

}
 

private void processNodes( List<Osmformat.Node> nodes, BPbfFieldDecoder fieldDecoder )
{
    for (Osmformat.Node node : nodes)
    {
        Map<String, String> tags = buildTags(node.getKeysList(), node.getValsList(), fieldDecoder);

        parser.addNode( node.getId(), tags, fieldDecoder.decodeLatitude(node
                .getLat()), fieldDecoder.decodeLatitude(node.getLon()));
    }
}
 

private LongList buildRelationMembers(
        List<Long> memberIds, List<Integer> memberRoles, List<Osmformat.Relation.MemberType> memberTypes,
        BPbfFieldDecoder fieldDecoder )
{
    LongList wayIds = new LongList( 16 );

    fromWid = toWid = viaNid = null;

    Iterator<Long> memberIdIterator = memberIds.iterator();
    Iterator<Integer> memberRoleIterator = memberRoles.iterator();
    Iterator<Osmformat.Relation.MemberType> memberTypeIterator = memberTypes.iterator();

    // Build up the list of relation members for the way. The member ids are
    // delta encoded meaning that each id is stored as a delta against
    // the previous one.
    long refId = 0;
    while (memberIdIterator.hasNext())
    {
        Osmformat.Relation.MemberType memberType = memberTypeIterator.next();
        refId += memberIdIterator.next();

        String role = fieldDecoder.decodeString( memberRoleIterator.next() );

        if ( memberType == Osmformat.Relation.MemberType.WAY ) // currently just waymembers
        {
          wayIds.add( refId );
          if ( "from".equals( role ) ) fromWid = addLong( fromWid, refId );
          if ( "to".equals( role ) ) toWid = addLong( toWid, refId );
        }
        if ( memberType == Osmformat.Relation.MemberType.NODE ) // currently just waymembers
        {
          if ( "via".equals( role ) ) viaNid = addLong( viaNid, refId );
        }
    }
    return wayIds;
}
 

private void processRelations( List<Osmformat.Relation> relations, BPbfFieldDecoder fieldDecoder )
{
    for (Osmformat.Relation relation : relations)
    {
        Map<String, String> tags = buildTags(relation.getKeysList(), relation.getValsList(), fieldDecoder);

        LongList wayIds = buildRelationMembers( relation.getMemidsList(), relation.getRolesSidList(),
                relation.getTypesList(), fieldDecoder);

        parser.addRelation( relation.getId(), tags, wayIds, fromWid, toWid, viaNid );
    }
}
 

private void processOsmPrimitives( byte[] data ) throws InvalidProtocolBufferException
{
    Osmformat.PrimitiveBlock block = Osmformat.PrimitiveBlock.parseFrom(data);
    BPbfFieldDecoder fieldDecoder = new BPbfFieldDecoder(block);

    for (Osmformat.PrimitiveGroup primitiveGroup : block.getPrimitivegroupList())
    {
        processNodes(primitiveGroup.getDense(), fieldDecoder);
        processNodes(primitiveGroup.getNodesList(), fieldDecoder);
        processWays(primitiveGroup.getWaysList(), fieldDecoder);
        processRelations(primitiveGroup.getRelationsList(), fieldDecoder);
    }
}
 

/**
 * Creates a new instance.
 * <p/>
 * @param primitiveBlock The primitive block containing the fields to be decoded.
 */
public BPbfFieldDecoder( Osmformat.PrimitiveBlock primitiveBlock )
{
    this.coordGranularity = primitiveBlock.getGranularity();
    this.coordLatitudeOffset = primitiveBlock.getLatOffset();
    this.coordLongitudeOffset = primitiveBlock.getLonOffset();
    this.dateGranularity = primitiveBlock.getDateGranularity();

    Osmformat.StringTable stringTable = primitiveBlock.getStringtable();
    strings = new String[stringTable.getSCount()];
    for (int i = 0; i < strings.length; i++)
    {
        strings[i] = stringTable.getS(i).toStringUtf8();
    }
}
 

@Override
protected void parseRelations(final List<Osmformat.Relation> rels) {
  for (final Osmformat.Relation r : rels) {
    final AvroRelation r2 = new AvroRelation();
    final AvroPrimitive p = getPrimitive(r.getInfo());
    p.setId(r.getId());
    p.setTags(getTags(r.getKeysList(), r.getValsList()));
    r2.setCommon(p);

    final List<AvroRelationMember> members = new ArrayList<>(r.getRolesSidCount());

    for (int i = 0; i < r.getRolesSidCount(); i++) {
      final AvroRelationMember rm = new AvroRelationMember();
      rm.setMember(r.getMemids(i));
      rm.setRole(getStringById(r.getRolesSid(i)));
      switch (r.getTypes(i).toString()) {
        case "NODE": {
          rm.setMemberType(AvroMemberType.NODE);
          break;
        }
        case "WAY": {
          rm.setMemberType(AvroMemberType.WAY);
          break;
        }
        case "RELATION": {
          rm.setMemberType(AvroMemberType.RELATION);
          break;
        }
        default:
          break;
      }
    }
    r2.setMembers(members);
    try {
      relationWriter.append(r2);
    } catch (final IOException e) {
      LOGGER.error("Unable to write relation", e);
    }
  }
}
 

@Override
protected void parseRelations(List<Osmformat.Relation> rels) {
    try {
        for (Osmformat.Relation r : rels) {
            if (relationCount++ % 100000 == 0) {
                LOG.info("relation {}", human(relationCount));
            }
            Relation rel = new Relation();
            /* Handle Tags */
            for (int k = 0; k < r.getKeysCount(); k++) {
                String key = getStringById(r.getKeys(k));
                String val = getStringById(r.getVals(k));
                if (retainTag(key)) rel.addTag(key, val);
            }
            /* Handle members of the relation */
            long mid = 0; // member ids, delta coded
            for (int m = 0; m < r.getMemidsCount(); m++) {
                Relation.Member member = new Relation.Member();
                mid += r.getMemids(m);
                member.id = mid;
                member.role = getStringById(r.getRolesSid(m));
                switch (r.getTypes(m)) {
                case NODE:
                    member.type = Type.NODE;
                    break;
                case WAY:
                    member.type = Type.WAY;
                    break;
                case RELATION:
                    member.type = Type.RELATION;
                    break;
                default:
                    LOG.error("Relation type is unexpected.");
                }
                rel.members.add(member);
            }
            entitySink.writeRelation(r.getId(), rel);
        }
    } catch (IOException ex) {
        LOG.error("An I/O exception occurred in the OSM entity sink.");
        ex.printStackTrace();
    }
}
 

private void processNodes( Osmformat.DenseNodes nodes, BPbfFieldDecoder fieldDecoder )
{
    List<Long> idList = nodes.getIdList();
    List<Long> latList = nodes.getLatList();
    List<Long> lonList = nodes.getLonList();

    Iterator<Integer> keysValuesIterator = nodes.getKeysValsList().iterator();

    long nodeId = 0;
    long latitude = 0;
    long longitude = 0;

    for (int i = 0; i < idList.size(); i++)
    {
        // Delta decode node fields.
        nodeId += idList.get(i);
        latitude += latList.get(i);
        longitude += lonList.get(i);

        // Build the tags. The key and value string indexes are sequential
        // in the same PBF array. Each set of tags is delimited by an index
        // with a value of 0.
        Map<String, String> tags = null;
        while (keysValuesIterator.hasNext())
        {
            int keyIndex = keysValuesIterator.next();
            if (keyIndex == 0)
            {
                break;
            }
            int valueIndex = keysValuesIterator.next();

            if (tags == null)
            {
                tags = new HashMap<String, String>();
            }

            tags.put(fieldDecoder.decodeString(keyIndex), fieldDecoder.decodeString(valueIndex));
        }

        parser.addNode( nodeId, tags, ((double) latitude) / 10000000, ((double) longitude) / 10000000);
    }
}
 

@Override
protected void parseDense(final Osmformat.DenseNodes nodes) {
  long lastId = 0;
  long lastLat = 0;
  long lastLon = 0;
  long lastTimestamp = 0;
  long lastChangeset = 0;
  int lastUid = 0;
  int lastSid = 0;

  int tagLocation = 0;

  for (int i = 0; i < nodes.getIdCount(); i++) {

    final AvroNode n = new AvroNode();
    final AvroPrimitive p = new AvroPrimitive();

    lastId += nodes.getId(i);
    lastLat += nodes.getLat(i);
    lastLon += nodes.getLon(i);

    p.setId(lastId);
    n.setLatitude(parseLat(lastLat));
    n.setLongitude(parseLon(lastLon));

    // Weird spec - keys and values are mashed sequentially, and end
    // of data for a particular node is denoted by a value of 0
    if (nodes.getKeysValsCount() > 0) {
      final Map<String, String> tags = new HashMap<>(nodes.getKeysValsCount());
      while (nodes.getKeysVals(tagLocation) > 0) {
        final String k = getStringById(nodes.getKeysVals(tagLocation));
        tagLocation++;
        final String v = getStringById(nodes.getKeysVals(tagLocation));
        tagLocation++;
        tags.put(k, v);
      }
      p.setTags(tags);
    }

    if (nodes.hasDenseinfo()) {
      final Osmformat.DenseInfo di = nodes.getDenseinfo();
      lastTimestamp += di.getTimestamp(i);
      lastChangeset += di.getChangeset(i);
      lastUid += di.getUid(i);
      lastSid += di.getUserSid(i);

      p.setTimestamp(lastTimestamp);
      p.setChangesetId(lastChangeset);
      p.setUserId((long) lastUid);
      p.setUserName(getStringById(lastSid));
      if (di.getVisibleCount() > 0) {
        p.setVisible(di.getVisible(i));
      }
    }

    n.setCommon(p);

    try {
      nodeWriter.append(n);
    } catch (final IOException e) {
      LOGGER.error("Unable to write dense node", e);
    }
  }
}
 

@Override
protected void parse(final Osmformat.HeaderBlock header) {}