Merge pull request #19 from HSLdevcom/develop

paasovaara · web-flow · commit 9b23bac75e05 · 2019-04-23T14:32:34.000+03:00
version 1.0.0
diff --git a/.travis.yml b/.travis.yml
@@ -7,29 +7,39 @@ jdk:
 services:
   - docker
 
+addons:
+  apt:
+    packages:
+      # Needed for `xmllint`.
+      - libxml2-utils
+
 env:
   global:
     # Docker Hub cannot handle uppercase characters in repository names. Fix it
     # with a bashism as Travis CI cannot handle quoting within command
     # substitution.
     - LOWERCASE_REPO_SLUG="${TRAVIS_REPO_SLUG,,}"
     - DOCKER_LOCAL="${LOWERCASE_REPO_SLUG}:${TRAVIS_COMMIT}"
-    - DOCKER_MASTER="${LOWERCASE_REPO_SLUG}:latest"
+    - DOCKER_DEV_TAG="${LOWERCASE_REPO_SLUG}:develop"
+    - BUILD_HASH=$(echo "${TRAVIS_COMMIT}" | cut -c1-7)
 
 before_install:
   - docker --version
   - echo "ENV GIT_COMMIT ${TRAVIS_COMMIT}" >> Dockerfile
 
 script:
+  - set -e
+  - ./.travis/check_semver
   - docker build --tag "${DOCKER_LOCAL}" .
 
+# develop-branch pushes the image with "develop"-tag. Tagged commit will push the image with fully versioned tags
 deploy:
   - provider: script
-    script: sh ./.travis/docker_login_tag_push "${DOCKER_LOCAL}" "${DOCKER_MASTER}"
+    script: sh ./.travis/docker_login_tag_push "${DOCKER_LOCAL}" "${DOCKER_DEV_TAG}"
     on:
-      branch: master
+      branch: develop
   - provider: script
-    script: sh ./.travis/docker_login_tag_semver_push "${DOCKER_LOCAL}" "${LOWERCASE_REPO_SLUG}" "${TRAVIS_TAG}"
+    script: sh ./.travis/docker_login_tag_semver_push "${DOCKER_LOCAL}" "${LOWERCASE_REPO_SLUG}" "${TRAVIS_TAG}" "${BUILD_HASH}"
     on:
       tags: true
       all_branches: true
diff --git a/.travis/check_semver b/.travis/check_semver
@@ -0,0 +1,34 @@
+#!/bin/bash
+
+matches_semver() {
+  # Simple regex-check, allows optionally three levels separated with .
+  if [[ "$1" =~ [0-9]+(\.[0-9]+(\.[0-9]+)?)?$ ]]; then
+    # 0 = true
+    return 0
+  else
+    # 1 = false
+    return 1
+  fi
+}
+
+set -u
+
+if [ -n "${TRAVIS_TAG}" ]; then
+  TRAVIS_SEMVER="$(matches_semver "${TRAVIS_TAG}")"
+  if [ "$?" -ne 0 ]; then
+    echo "TRAVIS_TAG ${TRAVIS_TAG} does not follow semver" 1>&2
+    exit 1
+  fi
+  PACKAGE_VERSION="$(xmllint --nsclean --xpath '/project/version/text()' pom.xml)"
+  PACKAGE_SEMVER="$(matches_semver "${PACKAGE_VERSION}")"
+  if [ "$?" -ne 0 ]; then
+    echo "version ${PACKAGE_VERSION} in pom.xml does not follow semver" 1>&2
+    exit 1
+  fi
+  if [ "${TRAVIS_TAG}" != "${PACKAGE_VERSION}" ]; then
+    echo "According to semver, TRAVIS_TAG ${TRAVIS_TAG} differs from version ${PACKAGE_VERSION} in pom.xml" 1>&2
+    exit 1
+  else
+    echo "Travis tag and pom.xml versions match"
+  fi
+fi
diff --git a/.travis/docker_login_tag_semver_push b/.travis/docker_login_tag_semver_push
@@ -2,14 +2,17 @@
 
 set -u
 
-[ "$#" -ne 3 ] && {
-  echo "Usage: $(basename "$0") SOURCE_IMAGE[:TAG] TARGET_IMAGE_WITHOUT_TAG TRAVIS_TAG" 1>&2
+[ "$#" -lt 3 ] && {
+  echo "Usage: $(basename "$0") SOURCE_IMAGE[:TAG] TARGET_IMAGE_WITHOUT_TAG TRAVIS_TAG [BUILD_HASH]" 1>&2
   exit 1
 }
 
 SOURCE_IMAGE_TAG="$1"
 TARGET_IMAGE_WITHOUT_TAG="$2"
 TAG_FROM_TRAVIS="$3"
+if [ "$#" -ge 4 ]; then
+  BUILD_HASH="$4"
+fi
 
 PUSH='./.travis/docker_login_tag_push'
 
@@ -27,6 +30,9 @@ if [ "${IS_PRE_RELEASE}" != true ]; then
   SEMVER_MAJOR="$(echo "${SEMVER_MINOR}" | cut -d . -f 1)"
   TARGET_IMAGES="${TARGET_IMAGES} ${TARGET_IMAGE_WITHOUT_TAG}:${SEMVER_MINOR}"
   TARGET_IMAGES="${TARGET_IMAGES} ${TARGET_IMAGE_WITHOUT_TAG}:${SEMVER_MAJOR}"
+  if [ -n "${BUILD_HASH-}" ]; then
+    TARGET_IMAGES="${TARGET_IMAGES} ${TARGET_IMAGE_WITHOUT_TAG}:${SEMVER_FULL}-${BUILD_HASH}"
+  fi
 fi
-
+echo "Pushing source image with tags $TARGET_IMAGES"
 "${PUSH}" "${SOURCE_IMAGE_TAG}" ${TARGET_IMAGES}
diff --git a/pom.xml b/pom.xml
@@ -2,7 +2,7 @@
     <modelVersion>4.0.0</modelVersion>
     <groupId>fi.hsl.transitdata</groupId>
     <artifactId>transitdata-pubtrans-source</artifactId>
-    <version>0.3.0</version>
+    <version>1.0.0</version>
     <packaging>jar</packaging>
 
     <repositories>
@@ -17,8 +17,7 @@
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
         <maven.compiler.source>1.8</maven.compiler.source>
         <maven.compiler.target>1.8</maven.compiler.target>
-        <pulsar.version>2.1.0-incubating</pulsar.version>
-        <common.version>0.4.0</common.version>
+        <common.version>1.0.0</common.version>
     </properties>
 
     <dependencies>
diff --git a/src/main/java/fi/hsl/transitdata/pulsarpubtransconnect/ArrivalHandler.java b/src/main/java/fi/hsl/transitdata/pulsarpubtransconnect/ArrivalHandler.java
@@ -2,6 +2,7 @@
 
 import fi.hsl.common.pulsar.PulsarApplicationContext;
 import fi.hsl.common.transitdata.TransitdataProperties;
+import fi.hsl.common.transitdata.TransitdataSchema;
 import fi.hsl.common.transitdata.proto.PubtransTableProtos;
 import org.apache.pulsar.client.api.Producer;
 import org.apache.pulsar.client.api.TypedMessageBuilder;
@@ -15,6 +16,12 @@
 
 public class ArrivalHandler extends PubtransTableHandler {
 
+    static final TransitdataSchema schema;
+    static {
+        int defaultVersion = PubtransTableProtos.ROIArrival.newBuilder().getSchemaVersion();
+        schema = new TransitdataSchema(TransitdataProperties.ProtobufSchema.PubtransRoiArrival, Optional.of(defaultVersion));
+    }
+
     public ArrivalHandler(PulsarApplicationContext context) {
         super(context, TransitdataProperties.ProtobufSchema.PubtransRoiArrival);
     }
@@ -25,9 +32,16 @@ protected String getTimetabledDateTimeColumnName() {
     }
 
     @Override
-    protected byte[] createPayload(ResultSet resultSet, PubtransTableProtos.Common common) throws SQLException {
+    protected TransitdataSchema getSchema() {
+        return schema;
+    }
+
+    @Override
+    protected byte[] createPayload(ResultSet resultSet, PubtransTableProtos.Common common, PubtransTableProtos.DOITripInfo tripInfo) throws SQLException {
         PubtransTableProtos.ROIArrival.Builder arrivalBuilder = PubtransTableProtos.ROIArrival.newBuilder();
+        arrivalBuilder.setSchemaVersion(arrivalBuilder.getSchemaVersion());
         arrivalBuilder.setCommon(common);
+        arrivalBuilder.setTripInfo(tripInfo);
         PubtransTableProtos.ROIArrival arrival = arrivalBuilder.build();
         return arrival.toByteArray();
     }
diff --git a/src/main/java/fi/hsl/transitdata/pulsarpubtransconnect/DepartureHandler.java b/src/main/java/fi/hsl/transitdata/pulsarpubtransconnect/DepartureHandler.java
@@ -2,6 +2,7 @@
 
 import fi.hsl.common.pulsar.PulsarApplicationContext;
 import fi.hsl.common.transitdata.TransitdataProperties;
+import fi.hsl.common.transitdata.TransitdataSchema;
 import fi.hsl.common.transitdata.proto.PubtransTableProtos;
 import org.apache.pulsar.client.api.Producer;
 import org.apache.pulsar.client.api.TypedMessageBuilder;
@@ -17,6 +18,12 @@
 
 public class DepartureHandler extends PubtransTableHandler {
 
+    static final TransitdataSchema schema;
+    static {
+        int defaultVersion = PubtransTableProtos.ROIDeparture.newBuilder().getSchemaVersion();
+        schema = new TransitdataSchema(TransitdataProperties.ProtobufSchema.PubtransRoiDeparture, Optional.of(defaultVersion));
+    }
+
     public DepartureHandler(PulsarApplicationContext context) {
         super(context, TransitdataProperties.ProtobufSchema.PubtransRoiDeparture);
     }
@@ -27,9 +34,16 @@ protected String getTimetabledDateTimeColumnName() {
     }
 
     @Override
-    protected byte[] createPayload(ResultSet resultSet, PubtransTableProtos.Common common) throws SQLException {
+    protected TransitdataSchema getSchema() {
+        return schema;
+    }
+
+    @Override
+    protected byte[] createPayload(ResultSet resultSet, PubtransTableProtos.Common common, PubtransTableProtos.DOITripInfo tripInfo) throws SQLException {
         PubtransTableProtos.ROIDeparture.Builder departureBuilder = PubtransTableProtos.ROIDeparture.newBuilder();
+        departureBuilder.setSchemaVersion(departureBuilder.getSchemaVersion());
         departureBuilder.setCommon(common);
+        departureBuilder.setTripInfo(tripInfo);
         if (resultSet.getBytes("HasDestinationDisplayId") != null)
             departureBuilder.setHasDestinationDisplayId(resultSet.getLong("HasDestinationDisplayId"));
         if (resultSet.getBytes("HasDestinationStopAreaGid") != null)
diff --git a/src/main/java/fi/hsl/transitdata/pulsarpubtransconnect/PubtransTableHandler.java b/src/main/java/fi/hsl/transitdata/pulsarpubtransconnect/PubtransTableHandler.java
@@ -2,6 +2,7 @@
 
 import fi.hsl.common.pulsar.PulsarApplicationContext;
 import fi.hsl.common.transitdata.TransitdataProperties;
+import fi.hsl.common.transitdata.TransitdataSchema;
 import fi.hsl.common.transitdata.proto.PubtransTableProtos;
 import org.apache.pulsar.client.api.Producer;
 import org.apache.pulsar.client.api.TypedMessageBuilder;
@@ -63,10 +64,12 @@ public static Optional<Long> toUtcEpochMs(String localTimestamp, String zoneId)
         }
     }
 
-    abstract protected byte[] createPayload(ResultSet resultSet, PubtransTableProtos.Common common) throws SQLException;
+    abstract protected byte[] createPayload(ResultSet resultSet, PubtransTableProtos.Common common, PubtransTableProtos.DOITripInfo tripInfo) throws SQLException;
 
     abstract protected String getTimetabledDateTimeColumnName();
 
+    abstract protected TransitdataSchema getSchema();
+
     public Queue<TypedMessageBuilder<byte[]>> handleResultSet(ResultSet resultSet) throws SQLException {
 
         Queue<TypedMessageBuilder<byte[]>> messageBuilderQueue = new LinkedList<>();
@@ -83,10 +86,16 @@ public Queue<TypedMessageBuilder<byte[]>> handleResultSet(ResultSet resultSet) t
             final String key = resultSet.getString("IsOnDatedVehicleJourneyId") + resultSet.getString("JourneyPatternSequenceNumber");
             final long dvjId = common.getIsOnDatedVehicleJourneyId();
             final long jppId = common.getIsTargetedAtJourneyPatternPointGid();
-            final byte[] data = createPayload(resultSet, common);
 
-            Optional<TypedMessageBuilder<byte[]>> maybeBuilder = createMessage(key, eventTimestampUtcMs, dvjId, jppId, data);
-            maybeBuilder.ifPresent(messageBuilderQueue::add);
+            Optional<PubtransTableProtos.DOITripInfo> maybeTripInfo = getTripInfo(dvjId, jppId);
+            if (!maybeTripInfo.isPresent()) {
+                log.warn("Could not find valid DOITripInfo from Redis for dvjId {}, jppId {}. Ignoring this update ", dvjId, jppId);
+            }
+            else {
+                final byte[] data = createPayload(resultSet, common, maybeTripInfo.get());
+                TypedMessageBuilder<byte[]> msgBuilder = createMessage(key, eventTimestampUtcMs, dvjId, data, getSchema());
+                messageBuilderQueue.add(msgBuilder);
+            }
 
             //Update latest ts for next round
             if (eventTimestampUtcMs > tempTimeStamp) {
@@ -133,77 +142,55 @@ protected PubtransTableProtos.Common parseCommon(ResultSet resultSet) throws SQL
         return commonBuilder.build();
     }
 
-
-    class JourneyInfo {
-        String direction;
-        String routeName;
-        String startTime;
-        String operatingDay;
-
-        JourneyInfo(Map<String, String> map) {
-            direction = map.get(TransitdataProperties.KEY_DIRECTION);
-            routeName = map.get(TransitdataProperties.KEY_ROUTE_NAME);
-            startTime = map.get(TransitdataProperties.KEY_START_TIME);
-            operatingDay = map.get(TransitdataProperties.KEY_OPERATING_DAY);
-            if (!isValid()) {
-                //Let's print more info for debugging purposes:
-                log.warn("JourneyInfo is missing some fields. Content: " + this.toString());
-            }
-        }
-
-        boolean isValid() {
-            return direction != null && routeName != null &&
-                   startTime != null && operatingDay != null;
-        }
-
-        @Override
-        public String toString() {
-            return new StringBuilder()
-                    .append("direction: ").append(direction)
-                    .append(" routeName: ").append(routeName)
-                    .append(" startTime: ").append(startTime)
-                    .append(" operatingDay: ").append(operatingDay)
-                    .toString();
-        }
-    }
-
-    private Optional<JourneyInfo> getJourneyInfo(long dvjId) {
-        String key = TransitdataProperties.REDIS_PREFIX_DVJ + Long.toString(dvjId);
-        Optional<Map<String, String>> maybeJourneyInfoMap = Optional.ofNullable(jedis.hgetAll(key));
-
-        return maybeJourneyInfoMap
-                .map(JourneyInfo::new)
-                .filter(JourneyInfo::isValid);
+    private Optional<String> getStopId(long jppId) {
+        String stopIdKey = TransitdataProperties.REDIS_PREFIX_JPP + Long.toString(jppId);
+        return Optional.ofNullable(jedis.get(stopIdKey));
     }
 
-    private Optional<String> getStopId(long jppId) {
-        String key = TransitdataProperties.REDIS_PREFIX_JPP + Long.toString(jppId);
-        return Optional.ofNullable(jedis.get(key));
+    private Optional<Map<String, String>> getTripInfoFields(long dvjId) {
+        String tripInfoKey = TransitdataProperties.REDIS_PREFIX_DVJ + Long.toString(dvjId);
+        return Optional.ofNullable(jedis.hgetAll(tripInfoKey));
     }
 
-    Optional<TypedMessageBuilder<byte[]>> createMessage(String key, long eventTime, long dvjId, long jppId, byte[] data) {
-        Optional<JourneyInfo> maybeJourneyInfo = getJourneyInfo(dvjId);
-        if (!maybeJourneyInfo.isPresent()) {
-            log.warn("Could not find valid JourneyInfo from Redis for dvjId " + dvjId);
+    protected Optional<PubtransTableProtos.DOITripInfo> getTripInfo(long dvjId, long jppId) {
+        try {
+            Optional<String> maybeStopId = getStopId(jppId);
+            Optional<Map<String, String>> maybeTripInfoMap = getTripInfoFields(dvjId);
+
+            if (maybeStopId.isPresent() && maybeTripInfoMap.isPresent()) {
+                PubtransTableProtos.DOITripInfo.Builder builder = PubtransTableProtos.DOITripInfo.newBuilder();
+                builder.setStopId(maybeStopId.get());
+                maybeTripInfoMap.ifPresent(map -> {
+                    if (map.containsKey(TransitdataProperties.KEY_DIRECTION))
+                        builder.setDirectionId(Integer.parseInt(map.get(TransitdataProperties.KEY_DIRECTION)));
+                    if (map.containsKey(TransitdataProperties.KEY_ROUTE_NAME))
+                        builder.setRouteId(map.get(TransitdataProperties.KEY_ROUTE_NAME));
+                    if (map.containsKey(TransitdataProperties.KEY_START_TIME))
+                        builder.setStartTime(map.get(TransitdataProperties.KEY_START_TIME));
+                    if (map.containsKey(TransitdataProperties.KEY_OPERATING_DAY))
+                        builder.setOperatingDay(map.get(TransitdataProperties.KEY_OPERATING_DAY));
+                });
+                builder.setDvjId(dvjId);
+                return Optional.of(builder.build());
+            }
+            else {
+                log.error("Failed to get data from Redis for dvjId {}, jppId {}", dvjId, jppId);
+                return Optional.empty();
+            }
         }
-        Optional<String> maybeStopId = getStopId(jppId);
-        if (!maybeStopId.isPresent()) {
-            log.warn("Could not find StopId from Redis for dvjId " + dvjId);
+        catch (Exception e) {
+            log.error("Failed to get Trip Info for dvj-id " + dvjId, e);
+            return Optional.empty();
         }
+    }
 
-        return maybeJourneyInfo.flatMap(journeyInfo ->
-                maybeStopId.map(stopId ->
-                    producer.newMessage()
-                            .key(key)
-                            .eventTime(eventTime)
-                            .property(TransitdataProperties.KEY_DVJ_ID, Long.toString(dvjId))
-                            .property(TransitdataProperties.KEY_PROTOBUF_SCHEMA, schema.toString())
-                            .property(TransitdataProperties.KEY_DIRECTION, journeyInfo.direction)
-                            .property(TransitdataProperties.KEY_ROUTE_NAME, journeyInfo.routeName)
-                            .property(TransitdataProperties.KEY_START_TIME, journeyInfo.startTime)
-                            .property(TransitdataProperties.KEY_OPERATING_DAY, journeyInfo.operatingDay)
-                            .property(TransitdataProperties.KEY_STOP_ID, stopId)
-                            .value(data))
-                );
+    protected TypedMessageBuilder<byte[]> createMessage(String key, long eventTime, long dvjId, byte[] data, TransitdataSchema schema) {
+        return producer.newMessage()
+                .key(key)
+                .eventTime(eventTime)
+                .property(TransitdataProperties.KEY_DVJ_ID, Long.toString(dvjId))
+                .property(TransitdataProperties.KEY_PROTOBUF_SCHEMA, schema.schema.toString())
+                .property(TransitdataProperties.KEY_SCHEMA_VERSION, Integer.toString(schema.schemaVersion.get()))
+                .value(data);
     }
 }
