For a Magik routine (method or procedure) to produce a stream for DTS, it must return only one result and it must be a stream-type object.
DTS considers any object that responds to close() and get_upto_n() to be a stream-type object.
While certain existing Magik constructs like record_stream, difference_stream and others do satisfy the stream-type object criteria, DTS also provides its own stream wrapper (dts:dts_record_stream) for generating streamable responses.
dts:dts_record_stream provides the following constructor methods:
A. For Wrapping Collections:
dts_record_stream.new_for_collection(
p_id,
p_collection,
_optional p_predicate)
•p_id: is ignored in when the method is used in this scenario;
•p_collection: the collection to stream from;
•p_predicate: a predicate to query the collection. If unset, the entire collection will be streamed.
Example:
_global stream_example1 <<
_proc @stream_example1 ()
_local collection << gis_program_manager.cached_dataset(:gis).collection(:min_road)
_return dts:dts_record_stream.new_for_collection(_unset, collection, predicate.eq(:road_type, "A-Road"))
_endproc
$
In this case, we would need to register a min_road DS type with DTS first, then register the routine call returning a stream of min_road types.
B. For Wrapping Other Streams:
dts_record_stream.new_for_stream(
p_stream)
•p_stream: a stream-type object
While it may seem somewhat redundant to wrap a stream-type object into a dts_record_stream, we recommend this to be done whenever feasible due to better compliance of the dts_record_stream objects with the DTS engine
Example:
_global stream_example2 <<
_proc @stream_example2 ()
_local stream << gis_program_manager.cached_dataset(:gis).collection(:min_road).read_stream()
_return dts:dts_record_stream.new_for_stream(stream)
_endproc
$
C. For Wrapping Vectors:
dts_record_stream.new_for_vector(
p_vector)
•p_vector: a Magik vector-type object (must respond to subseq() and size) to be wrapped into a stream.
Example:
_global stream_example3 <<
_proc @stream_example3 (p_size)
_local vector << rope.new()
_for i_index _over 1.upto(p_size)
_loop
vector.add(some_object.new(i_index))
_endloop
_return dts:dts_record_stream.new_for_vector(vector)
_endproc
$
In this case, we would to register an exemplar type for some_object with DTS, then register the routine call returning a stream of some_object types.