Sending multicardinal data in a custom adapter

Question

Hello,

I have a workflow which calculates the traffic state. For each street (segment) it receives sensors data. I have a custom processor which needs all sensors data in order to compute the segments states.

For each sensor i have the following data : sensorID, speed, occupancy_rate, traffic_flow

In my custom adapter, if i send a Geoevent message for each sensor, then it's difficult for the processor to know when it has received the data for all sensors. So I would like to send the sensors data in one message. Is it possible ?

I found examples how to define multi-cardinality fields and groups, but not how to send the data in a custom adapter. Could you help me how to do this ?

And also in the processor, how do you receive the multi-cardinal data ?

Regards,

Philippe De Pol.

RJSunderman · Answer

Philip --

Your solution using an outbound connector which is essentially an No Operation component is a bit orthogonal  to GeoEvent Server's design. What you're doing is one reason that we don't offer out-of-the-box processors with capabilities to invoke a GP Service, for example. We certainly could, as a GP service is as RESTful as other web services GeoEvent Server is interfacing with. But the question becomes do we want to block a processor node's flow as it waits on a response from a GP Service? This wouldn't be feasible when trying to process hundreds of event messages per second. Or do we allow the processor to invoke an asynchronous GP service task/job and have the processor send the logical equivalent of "nothing" or "process pending" along to an outbound connector? That's not consistent with GeoEvent Server's design.

GeoEvent Server is fundamentally accepting data, adapting the data to produce individual event records, then processing each of those event records atomically (without retaining or caching data from an event record unless absolutely necessary), so that data from a processed event record can be routed along to an outbound connector for dissemination.

Your solution appears to be developer-centric and highly customized. If I understand what you're saying you have a custom inbound adapter, a custom processor, and now a custom outbound connector. If the GeoEvent Server's Java SDK allows you to develop a solution using GeoEvent Server as a platform for event record processing -- that's great -- but I'm not sure that the product team can be of much help moving forward.

I will offer that the multicardinality and hierarchy supported by a GeoEvent Definition is not specific to JSON. How data is ingested and adapted is not tied to a specific data format (e.g. JSON object format). Every event record has a GeoEvent Definition which describes the event record's data structure. This event definition applies only to the interior of an event record object. There is no mechanism which allows you to define a group, list, or hierarchy of multiple event records.

A GeoEvent Definition can specify a data structure which includes a list of Java primitive values (e.g. Date, Double, Long, String, ...) and/or incorporate a non-primitive type Group which includes multiple primitive values as a sub-structure within the overall data structure. But this all sill describes the data structure of a single event record object. The hierarchy and multicardinality concepts discussed in the article you found do not apply to collections of multiple event records.

PhilippeDePol · Answer

Thanks a lot for your answer. I appreciate your effort.

For the moment i use in the processor a counter which is used to see if we have received all the data. Every time i receive a data, the counter is increased by one. The problem is then to update the datasource with a standard output connector. I don't think it's possible.

When the processor has received all the sensors data, it calculates the segments state, but then it's not possible to send in the output several records at the same time. So i use what it seems to me a dirty solution : in the processor, i use the feature service of the datasource  with an 'add query' to add all the records at the same time. As Geovent needs always an output connector in a workflow, i created an custom 'empty output connector', which is doing nothing.

Still, in the Geoevent developer guide and in the sample-adapter-custom-geoeventdefinition example, i saw it's possible to define group fields and fields with cardinality with the value 'many'. I was wondering if i could use these features, but I didn't find an example anywhere. I read also the API documentation but i didn't find how to do it with Geoevent object.

Also i read you article 'Json Data Structures - Working with Hierarchy and Multicardinality'. It seems to do more or less what i want, but i am not dealing with Json data.

RJSunderman · Answer

Hey Philip --

GeoEvent Server's processing of event data was designed to be atomic. Every event record is processed individually. Generally speaking a processor does not know anything about event records recently processed or event records in the pipeline about to be processed. It only knows what data is in the event record is has received that needs to be processed.

There are exceptions, of course. A filter or processor needing to evaluate a Enter condition for example needs to know if the previous event for a given tracked asset (identified using the TRACK_ID tag) was "outside" or "disjoint" so that it can determine that the event record it just received, which is "inside" or "intersects" has entered the area of interest.

You might look at the Timetree processor, a custom processor whose source code is available in a GitHub repository here, as an example of a custom processor designed to collect and cache a number of event records in order to perform some processing on a collection of received data records. But as you say, you'll have to design some sort of parameterization so the processor knows when to stop collecting data and start processing the collection.

I don't think it's possible to configure a GeoEvent Definition such that the data structure represents an amalgamation of multiple event records. Since every event record must have an associated GeoEvent Definition specifying the event record's data structure - I don't think you'll be able to do what you're asking. But I'll check with a colleague and reply back if it turns out this is possible and something reasonably accomplished.

-- RJ