Combined minor changes and fixes

common/networking/monitoring/README.md

-Currently changes to these files have to be compiled seperately for Go and JavaScript
+Currently, changes to these files have to be compiled separately for Go and JavaScript
 
 Execute each `./generate-proto.sh` in:
 - `#/monitoring/monitoring_ui`

examples/producer/dummy-data-producer/dummy_data_producer.cpp

@@ -41,7 +41,7 @@ void PrintCommandArguments(const Args& args) {
             << "Raw: " << (args.mode / 100 == 1) << std::endl
             << "timeout: " << args.timeout_ms << std::endl
             << "messages in set: " << args.messages_in_set << std::endl
-            << "pipeline: " << args.pipeline_name << std::endl
+            << "pipelineStep: " << args.pipeline_name << std::endl
               << std::endl;
 }
 

monitoring/monitoring_server/src/asapo_monitoring_server/server/QueryServer.go

@@ -451,7 +451,8 @@ func (s *QueryServer) GetTopology(ctx context.Context, query *pb.ToplogyQuery) (
 			var step = getOrCreateStep(stepId)
 			step.producesSourceId[source] = true
 			step.producerInstances[producerInstanceId] = true
-			step.involvedReceivers[result.Record().Values()["receiverName"].(string)] = true
+			var receiverName = result.Record().Values()["receiverName"].(string)
+			step.involvedReceivers[receiverName] = true
 		} else if result.Record().Values()["brokerName"] != nil { // data is coming from broker => means it must be a consumer
 			stepId := result.Record().Values()["pipelineStepId"].(string)
 			source := result.Record().Values()["source"].(string)
@@ -539,11 +540,12 @@ func (s *QueryServer) GetTopology(ctx context.Context, query *pb.ToplogyQuery) (
 						})
 					}
 					for _, sourceIdAvailableFromStepId := range availableSources[consumingSourceId] { // Add all the others edges
+						var producingSource = pipelineSteps[sourceIdAvailableFromStepId]
 						edges = append(edges, PipelineEdgeInfo{
 							fromStepId:        sourceIdAvailableFromStepId,
 							toStepId:          stepId,
 							sourceId:          consumingSourceId,
-							involvedReceivers: stepInfo.involvedReceivers,
+							involvedReceivers: producingSource.involvedReceivers,
 						})
 					}
 				}
@@ -567,8 +569,44 @@ func (s *QueryServer) GetTopology(ctx context.Context, query *pb.ToplogyQuery) (
 
 		if !foundAtLeastOne {
 			// probably only requests of files came, but no receiver registered a producer
-			log.Error("infinite loop while building topology tree; Still has pipeline steps but found no way how to connect them")
-			return nil, errors.New("infinite loop while building topology tree; Still has pipeline steps but found no way how to connect them")
+			// log.Error("infinite loop while building topology tree; Still has pipeline steps but found no way how to connect them")
+			// return nil, errors.New("infinite loop while building topology tree; Still has pipeline steps but found no way how to connect them")
+
+			var unkStep = "Unknown"
+
+			pipelineSteps[unkStep] = PipelineStep{
+				producesSourceId:  nil,
+				consumesSourceId:  nil,
+				producerInstances: nil,
+				consumerInstances: nil,
+				involvedReceivers: nil,
+			}
+
+			levels[0] = append(levels[0], PipelineLevelStepInfo{
+				stepId: unkStep,
+			})
+
+			if len(levels) < 2 {
+				levels = append(levels, []PipelineLevelStepInfo{})
+			}
+
+			for stepId := range remainingStepIds {
+				var step = pipelineSteps[stepId]
+				levels[1] = append(levels[1], PipelineLevelStepInfo{
+					stepId: stepId,
+				})
+
+				for sourceId := range step.consumesSourceId {
+					edges = append(edges, PipelineEdgeInfo{
+						fromStepId: unkStep,
+						toStepId:   stepId,
+						sourceId:   sourceId,
+						involvedReceivers: nil,
+					})
+				}
+			}
+
+			break // Go to response building
 		}
 
 		for sourceId, element := range availableSourcesInNextLevel {
@@ -582,6 +620,7 @@ func (s *QueryServer) GetTopology(ctx context.Context, query *pb.ToplogyQuery) (
 		}
 	}
 
+	// Response building
 	var response pb.TopologyResponse
 
 	for levelIndex, level := range levels {

monitoring/monitoring_ui/src/components/FilterSelector.vue

@@ -52,7 +52,7 @@
                     <select name="Source" id="source" class="w-32" v-model="selectedSource">
                         <option v-for="source in availableSources" :key="source">{{source}}</option>
                     </select>
-                    <button>X</button>
+                    <button @click="clearSourceFilter()">X</button>
                 </div>
             </div>
             <div class="mt-3 flex flex-col">
@@ -87,14 +87,23 @@ import { toplogyStore } from "../store/toplogyStore";
 
 @Options({
     watch: {
+        selectedBeamtimeFromStore: {
+            handler(newValue: string | null): void {
+                this.selectedBeamtime = newValue;
+            },
+            immediate: true,
+        },
         selectedBeamtime(newValue: string | null): void {
             selectionFilterStore.setFilterBeamtime(newValue);
         },
         currentBeamtimeFilterText(): void {
             this.selectedBeamtime = selectionFilterStore.state.beamtime;
         },
+        selectedSourceFromStore(newValue: string | null): void {
+            this.selectedSource = newValue;
+        },
         selectedSource(newValue: string | null): void {
-            //selectionFilterStore.setFilterBeamtime(newValue);
+            selectionFilterStore.setFilterSource(newValue);
         },
         currentSourceFilterText(): void {
             this.selectedSource = selectionFilterStore.state.source;
@@ -106,6 +115,13 @@ export default class FilterSelector extends Vue {
     private selectedSource: string | null = null;
     private showPopup: boolean = false;
 
+    private get selectedBeamtimeFromStore(): string | null {
+        return selectionFilterStore.state.beamtime;
+    }
+    private get selectedSourceFromStore(): string | null {
+        return selectionFilterStore.state.source;
+    }
+
     private get hasClearableFilter(): boolean {
         return selectionFilterStore.hasClearableFilter;
     }
@@ -152,7 +168,6 @@ export default class FilterSelector extends Vue {
     }
 
     private get hasFilterError(): boolean {
-        console.log('Filte error text: ', errorStore.state.filterErrorText)
         return !!errorStore.state.filterErrorText;
     }
 
@@ -167,6 +182,10 @@ export default class FilterSelector extends Vue {
         selectionFilterStore.setFilterBeamtime(null);
     }
 
+    private clearSourceFilter(): void {
+        selectionFilterStore.clearSourceFilter();
+    }
+
     private get currentSourceFilterText(): string {
         if (selectionFilterStore.state.source) {
             return selectionFilterStore.state.source;

monitoring/monitoring_ui/src/store/selectionFilterStore.ts

@@ -49,6 +49,26 @@ class SelectionFilterStore {
         }
     }
 
+    public clearSourceFilter(): void {
+        this.internalState.source = null;
+        this.internalState.fromPipelineStepId = null;
+        this.internalState.toPipelineStepId = null;
+        this.internalState.pipelineFilterRelation = 'and';
+        this.internalState.receiverId = null;
+    }
+
+    public setFilterSource(source: string | null): void {
+        if (source == null) {
+            this.clearSourceFilter();
+            return;
+        }
+        this.internalState.source = source;
+        this.internalState.fromPipelineStepId = null;
+        this.internalState.toPipelineStepId = null;
+        this.internalState.pipelineFilterRelation = 'or';
+        this.internalState.receiverId = null;
+    }
+
     public setFilterSourceWithPipeline(source: string, fromPipelineStepId: string, toPipelineStepId: string): void {
         this.internalState.source = source;
         this.internalState.fromPipelineStepId = fromPipelineStepId;

receiver/src/monitoring/receiver_monitoring_client_impl.cpp

@@ -221,6 +221,8 @@ asapo::Error asapo::ReceiverMonitoringClientImpl::ReinitializeClient() {
         newChannel.reset();
     }
 
+    StartSendingThread();
+
     return nullptr;
 }