277 pattern in redbull trains

anubis/src/components/DelaysTableView.vue

@@ -7,14 +7,15 @@
       type="text"
       placeholder="Search.."
     />
-    <grid-table :grid-data="receivedUpdates"></grid-table>
+    <grid-table id="table" :grid-data="receivedUpdates"></grid-table>
   </div>
 </template>
 
 <script lang="ts">
 import BasicTable from "../components/BasicTable.vue";
 import GridTable from "../components/GridTable.vue";
 import { AppModule } from "../store/modules/app";
+import store from "@/store";
 import { ReplayerModule } from "../store/modules/replayer";
 import { Component, Vue, Prop } from "vue-property-decorator";
 import { Notification } from "../generated/protobuf/models/internal/notifications/notification_pb";
@@ -31,39 +32,39 @@ export default class DelaysTableView extends Vue {
   protected search!: string;
 
   get receivedUpdates(): { [key: string]: any }[] {
+    return this.$store.state.notificationList.map(this.mapDelayToStringKey);
+
+    /*
     return AppModule.notifications
       .slice(
         AppModule.notifications.length - 51,
         AppModule.notifications.length - 1
       )
       .map(mapDelayToStringKey);
+
+     */
   }
-}
 
-function mapDelayToStringKey(
-  notification: Notification
-): { [key: string]: any } {
-  let delayStringKey: { [key: string]: any } = {
-    CeptaStationID: notification.getDelay()?.getStationId(),
-    CeptaID: notification.getDelay()?.getTransportId(),
-    Delay: notification
-      .getDelay()
-      ?.getDelay()
-      ?.getDelta(),
-    Details: notification
-      .getDelay()
-      ?.getDelay()
-      ?.getDetails()
-  };
+  mapDelayToStringKey(notification: Notification): { [key: string]: any } {
+    let delayStringKey: { [key: string]: any } = {
+      CeptaStationID: notification.getDelay()?.getStationId(),
+      CeptaID: notification.getDelay()?.getTransportId(),
+      Delay: notification
+        .getDelay()
+        ?.getDelay()
+        ?.getDelta(),
+      Details: notification
+        .getDelay()
+        ?.getDelay()
+        ?.getDetails()
+    };
 
-  return delayStringKey;
+    return delayStringKey;
+  }
 }
 </script>
 
 <style scoped lang="sass">
-#mainContent
-  padding: 0
-
 .form-control
   width: 100%
   margin-bottom: 20px

anubis/src/components/Sidebar.vue

@@ -239,11 +239,10 @@ export default class Sidebar extends Vue {
         display: inline-block
         top: 20%
         height: 60%
-        padding-right: 18px
 
     .logo-text
       +theme(color, c-logo-text)
-      padding-left: 10px
+      padding-left: 23px
 
   .mobile-toggle
     display: none

anubis/src/store/index.ts

@@ -1,5 +1,6 @@
 import Vue from "vue";
 import Vuex, { StoreOptions } from "vuex";
+import { Notification } from "../generated/protobuf/models/internal/notifications/notification_pb";
 import { IAppState } from "./modules/app";
 import { IAuthState } from "./modules/auth";
 import { IReplayerState } from "./modules/replayer";
@@ -16,11 +17,19 @@ export interface IRootState {
 // export default new Vuex.Store<IRootState>({
 export default new Vuex.Store({
   state: {
-    websocket: ""
+    notificationList: Array<Notification>()
   },
   mutations: {
-    setWebsocket(state: { websocket: String }, websocket: String) {
-      state.websocket = websocket;
+    addNotification(state, notification: Notification) {
+      if (state.notificationList.length > 100) {
+        state.notificationList = state.notificationList.slice(-99, -1);
+      }
+      state.notificationList = [notification, ...state.notificationList];
+    }
+  },
+  actions: {
+    addNotification(context, notification: Notification) {
+      context.commit("addNotification", notification);
     }
   }
 });

anubis/src/store/modules/app.ts

@@ -54,11 +54,6 @@ class App extends VuexModule implements IAppState {
     this.isLoading = loading;
   }
 
-  @Mutation
-  public addNotification(notification: Notification) {
-    this.notifications.push(notification);
-  }
-
   @Mutation
   public toggleTheme() {
     this.theme = (this.theme + 1) % this.availableThemes.length;

anubis/src/store/modules/notifications.ts

@@ -48,7 +48,7 @@ class Notifications extends VuexModule implements NotificationsState {
     let deserializedEvent = Notification.deserializeBinary(
       new Uint8Array(event.data)
     );
-    AppModule.addNotification(deserializedEvent);
+    store.dispatch("addNotification", deserializedEvent);
   }
 
   @Mutation

anubis/src/views/Dashboard.vue

@@ -50,7 +50,7 @@
         layoutStyle="col-md-12"
       >
         <p>You can use the navigation bar to filter</p>
-        <delays-table-view></delays-table-view>
+        <delays-table-view class="delay-table"></delays-table-view>
         <div class="view-all">
           <router-link :to="{ name: 'map' }">
             <div class="btn">
@@ -134,4 +134,10 @@ export default class Dashboard extends Vue {}
 
       &.metric-value
         font-size: 2.5rem
+
+
+  .delay-table
+    padding-right: 10px
+    height: 500px
+    overflow-y: scroll !important
 </style>

core/src/main/java/org/bptlab/cepta/Main.java

@@ -93,16 +93,16 @@ public class Main implements Callable<Integer> {
   private FlinkKafkaProducer011<NotificationOuterClass.Notification> trainDelayNotificationProducer;
 
   private void setupConsumers() {
-    this.liveTrainDataConsumer =
-        new FlinkKafkaConsumer011<>(
-          Topic.LIVE_TRAIN_DATA.getValueDescriptor().getName(),
-          new GenericBinaryProtoDeserializer<EventOuterClass.Event>(EventOuterClass.Event.class),
-          new KafkaConfig().withClientId("LiveTrainDataMainConsumer").getProperties());
-    this.plannedTrainDataConsumer =
-        new FlinkKafkaConsumer011<>(
-          Topic.PLANNED_TRAIN_DATA.getValueDescriptor().getName(),
-            new GenericBinaryProtoDeserializer<EventOuterClass.Event>(EventOuterClass.Event.class),
-            new KafkaConfig().withClientId("PlannedTrainDataMainConsumer").getProperties());
+      this.liveTrainDataConsumer =
+          new FlinkKafkaConsumer011<>(
+              Topic.LIVE_TRAIN_DATA.getValueDescriptor().getName(),
+              new GenericBinaryProtoDeserializer<EventOuterClass.Event>(EventOuterClass.Event.class),
+              new KafkaConfig().withClientId("LiveTrainDataMainConsumer").getProperties());
+      this.plannedTrainDataConsumer =
+          new FlinkKafkaConsumer011<>(
+              Topic.PLANNED_TRAIN_DATA.getValueDescriptor().getName(),
+              new GenericBinaryProtoDeserializer<EventOuterClass.Event>(EventOuterClass.Event.class),
+              new KafkaConfig().withClientId("PlannedTrainDataMainConsumer").getProperties());
 
     this.weatherDataConsumer =
         new FlinkKafkaConsumer011<>(

deployment/dev/compose/core.compose.yml

@@ -152,12 +152,13 @@ services:
       REDIS_HOST: redis
       REDIS_PORT: 6379
       LOG: DEBUG
+      TESTING: "true"
     ports:
       - ${CEPTA_NOTIFICATION_WS_PORT}:5000
     depends_on:
       #- flink-jobmanager
       #- flink-taskmanager
-      - kafka
+      #- kafkas
       #- replayer
       - redis
       - usermgmt

osiris/notification/server.go

@@ -3,8 +3,6 @@ package main
 import (
 	"context"
 	"fmt"
-	"github.com/bptlab/cepta/osiris/lib/utils"
-	"github.com/pkg/errors"
 	"io"
 	"net"
 	"net/http"
@@ -15,7 +13,9 @@ import (
 	"syscall"
 	"time"
 
-	"github.com/go-redis/redis"
+	"github.com/bptlab/cepta/osiris/lib/utils"
+	"github.com/pkg/errors"
+
 	"github.com/bptlab/cepta/ci/versioning"
 	topics "github.com/bptlab/cepta/models/constants/topic"
 	pb "github.com/bptlab/cepta/models/grpc/notification"
@@ -25,9 +25,10 @@ import (
 	"github.com/bptlab/cepta/models/internal/types/result"
 	"github.com/bptlab/cepta/models/internal/types/users"
 	libcli "github.com/bptlab/cepta/osiris/lib/cli"
-	libredis "github.com/bptlab/cepta/osiris/lib/redis"
 	kafkaconsumer "github.com/bptlab/cepta/osiris/lib/kafka/consumer"
+	libredis "github.com/bptlab/cepta/osiris/lib/redis"
 	"github.com/bptlab/cepta/osiris/notification/websocket"
+	"github.com/go-redis/redis"
 	clivalues "github.com/romnnn/flags4urfavecli/values"
 
 	"github.com/golang/protobuf/proto"
@@ -43,11 +44,13 @@ var Version string = "Unknown"
 // BuildTime will be injected at build time
 var BuildTime string = ""
 
+var testing bool
+
 var (
-	defaultLruSize                  = 1000 // Cache up to 1000 transports
-	defaultLruMaxEntryLength        = 1000 // Cache up to 1000 subscribers per transport
-	defaultUserNotificationsBufferSize        = int64(200) // Store the most recent 200 notifications for each user (not persistent)
-	defaultNotificationTopic = topics.Topic_DELAY_NOTIFICATIONS
+	defaultLruSize                     = 1000       // Cache up to 1000 transports
+	defaultLruMaxEntryLength           = 1000       // Cache up to 1000 subscribers per transport
+	defaultUserNotificationsBufferSize = int64(200) // Store the most recent 200 notifications for each user (not persistent)
+	defaultNotificationTopic           = topics.Topic_DELAY_NOTIFICATIONS
 )
 
 // Endpoint ...
@@ -69,7 +72,7 @@ type NotificationServer struct {
 	transportCache *lru.Cache
 	Pool           *websocket.Pool
 
-	usermgmtConn *grpc.ClientConn
+	usermgmtConn   *grpc.ClientConn
 	usermgmtClient usermgmtpb.UserManagementClient
 	grpcServer     *grpc.Server
 	wsServer       *http.Server
@@ -78,26 +81,26 @@ type NotificationServer struct {
 	kc           *kafkaconsumer.Consumer
 
 	redisConfig libredis.Config
-	rclient *redis.Client
+	rclient     *redis.Client
 
 	usermgmtEndpoint Endpoint
 
-	LruSize int
-	LruMaxEntryLength int
+	LruSize                     int
+	LruMaxEntryLength           int
 	UserNotificationsBufferSize int64
-	NotificationTopic topics.Topic
+	NotificationTopic           topics.Topic
 }
 
 // NewNotificationServer ...
 func NewNotificationServer(kafkaConfig kafkaconsumer.Config, redisConfig libredis.Config) NotificationServer {
 	return NotificationServer{
 		kafkacConfig: kafkaConfig,
-		redisConfig: redisConfig,
+		redisConfig:  redisConfig,
 		// Use defaults
-		LruSize: defaultLruSize,
-		LruMaxEntryLength: defaultLruMaxEntryLength,
+		LruSize:                     defaultLruSize,
+		LruMaxEntryLength:           defaultLruMaxEntryLength,
 		UserNotificationsBufferSize: defaultUserNotificationsBufferSize,
-		NotificationTopic: defaultNotificationTopic,
+		NotificationTopic:           defaultNotificationTopic,
 	}
 }
 
@@ -118,7 +121,7 @@ func (s *NotificationServer) Shutdown() {
 	}
 	if s.wsServer != nil {
 		log.Info("Stopping websocket server")
-		if err := s.wsServer. Shutdown(context.TODO()); err != nil {
+		if err := s.wsServer.Shutdown(context.TODO()); err != nil {
 			log.Warnf("Failed to close websocket server cleanly: %v", err)
 		}
 	}
@@ -274,18 +277,26 @@ func (s *NotificationServer) serveWebsocket(pool *websocket.Pool, w http.Respons
 	}
 
 	pool.Register <- client
-	client.Read()
 }
 
 func (s *NotificationServer) handleKafkaMessages(ctx context.Context) {
-	noopTicker := time.NewTicker(time.Second * 10)
+	// Send pings to client with this period. Must be less than pongWait on the client.
+	const pingPeriod = time.Second * 2
+	noopTicker := time.NewTicker(pingPeriod)
 	subscriberDone := make(chan bool, 1)
 	stopSubscriber := make(chan bool, 1)
 	go func() {
-		defer func() { subscriberDone <- true }()
+		defer func() {
+			stopSubscriber <- true
+			noopTicker.Stop()
+		}()
 		for {
 			select {
-
+			case <-noopTicker.C:
+				// Noop, may be used for periodic pings
+				log.Debug("ping")
+				s.Pool.Ping <- "ping"
+				break
 			case msg := <-s.kc.Messages:
 				var notification notificationpb.Notification
 				err := proto.Unmarshal(msg.Value, &notification)
@@ -295,7 +306,7 @@ func (s *NotificationServer) handleKafkaMessages(ctx context.Context) {
 				log.Debugf("Received notification: %v (%v)", notification, msg.Timestamp)
 
 				// Set occurrence time as a best effort
-				if occurred := notification.GetOccurred(); occurred.GetSeconds() + int64(occurred.GetNanos()) < 1 && !msg.Timestamp.IsZero() {
+				if occurred := notification.GetOccurred(); occurred.GetSeconds()+int64(occurred.GetNanos()) < 1 && !msg.Timestamp.IsZero() {
 					if occurredProto, err := utils.ToProtoTime(msg.Timestamp); err != nil {
 						notification.Occurred = occurredProto
 					}
@@ -304,20 +315,20 @@ func (s *NotificationServer) handleKafkaMessages(ctx context.Context) {
 				switch notification.GetNotification().(type) {
 				case *notificationpb.Notification_Delay:
 					// Just for testing porpuses
-					// s.broadcast(&notification)
-					// Users need to be assigned to transports to do this
-
-					if err := s.notifySubscribersForTransport(ctx, notification.GetDelay().GetTransportId(), &notification); err != nil {
-						log.Errorf("Failed to notify subscribers of transport %v: %v", notification.GetDelay().GetTransportId(), err)
+					if testing {
+						s.broadcast(&notification)
+					} else {
+						// Users need to be assigned to transports to do this
+						if err := s.notifySubscribersForTransport(ctx, notification.GetDelay().GetTransportId(), &notification); err != nil {
+							log.Errorf("Failed to notify subscribers of transport %v: %v", notification.GetDelay().GetTransportId(), err)
+						}
 					}
 					break
 				case *notificationpb.Notification_System:
 					s.broadcast(&notification)
 					break
 				}
 				break
-			case <-noopTicker.C:
-				// Noop, may be used for periodic pings
 			case <-stopSubscriber:
 				return
 			}
@@ -458,6 +469,12 @@ func main() {
 			EnvVars: []string{"NOTIFICATIONS_TOPIC"},
 			Usage:   "topic to subscribe for notifications",
 		},
+		&cli.BoolFlag{
+			Name:    "testing",
+			Value:   false,
+			EnvVars: []string{"TESTING"},
+			Usage:   "notification service broadcasts all messages for testing purposes",
+		},
 	}...)
 
 	app := &cli.App{
@@ -472,6 +489,7 @@ func main() {
 				level = log.InfoLevel
 			}
 			log.SetLevel(level)
+			testing = ctx.Bool("testing")
 
 			server := NewNotificationServer(kafkaconsumer.Config{}.ParseCli(ctx), libredis.Config{}.ParseCli(ctx))
 			server.usermgmtEndpoint = Endpoint{Host: ctx.String("usermgmt-host"), Port: ctx.Int("usermgmt-port")}