Reactive Stream Processing Using DDS and Rx

Your systems. Working as one. Reactive Stream Processingusing DDSand Rx www.rti.com Sumant Tambe, Ph.D. Senior Software Research Engineer and Microsoft MVP Real-Time Innovations, Inc. @sutambe Oct. 11, 2014

Outline •Reactive Systems •Stream Processing •Overview of Reactive Extensions •Overview of DDS •Streaming Shapes Demo in C# 10/10/2014 Real-Time Innovations, Inc. 2

Systems Everyone Wants to Build •Event-Driven:Modular, pipelined, asynchronous •Elasic:Scales easily up/down with load and cpucores. •Resilient:fault-tolerant •Responsive:Reacts to events at the speed of environment 10/10/2014 Real-Time Innovations, Inc. 3 Networking Middleware App

Stream Processing •Stream Processing is the term used to describe an architectural style that operate on a continuous sequence of data. 10/10/2014 Real-Time Innovations, Inc. 4

Shape of an application o/p Where Once CombineLatest Select Scan Merge Raw Data i/p

*nix command line (pipes and filter) $ ls-1 | grep“search” | grep“research” research $ 10/10/2014 Real-Time Innovations, Inc. 6 $ cat -| grep“Real” R Real Real [Ctrl+C] $

Reactive Extensions •The Reactive Extensions (Rx) is a library for composing asynchronous and event-based programs using observablesequences •Rx = Observables + Composition + Schedulers •Streams are first-class •Filter, project, aggregate, compose and perform time-based operations on multiple streams •Uses Functional Programming Style •Rx.NET, RxJava, RxJS, RxCpp, RxRuby, RxPyton, and more… 10/10/2014 Real-Time Innovations, Inc. 7

10/10/2014 Real-Time Innovations, Inc. 8 Data Distribution Service (DDS) A Reactive Middleware

Systems that interact with the Real World •Must adapt to changing environment •Cannot stop processing the information •Live within world-imposed timing Beyond traditional interpretation of real-time © 2010 Real-Time Innovations, Inc.

RPC over DDS 2014 DDS Security 2012 Family of Specifications © 2009 Real-Time Innovations, Inc. COMPANY 12 DDS Implementation Network / TCP / UDP / IP App DDS Implementation App DDS Implementation DDS Spec 2004 DDS Interoperablity 2006 UML Profile for DDS 2008 DDS for Lw CCM 2009 DDS X-Types 2010 2010 DDS-STD-C++ DDS-JAVA5 App

Evolution of DataBus Data-centricity basics

Everyday Example: Schedule Meeting via Emails Alternative Process #1 (message-centric): 1.Email: “Meeting Monday at 10:00.” 2.Email: “Here’s dial-in info for meeting…” 3.Email: “Meeting moved to Tuesday” 4.You: “Where do I have to be? When?” 5.You: (sifting through email messages…) 14

Everyday Example: Schedule Meeting Using a Calendar Alternative Process #2: 1.Calendar: (add meeting Monday at 10:00) 2.Calendar: (add dial-in info) 3.Calendar: (move meeting to Tuesday) 4.You: “Where do I have to be? When?” 5.You: (check calendar. Contains consolidated-state) 15 The difference is state! The infrastructure consolidates changes and maintains it

DDS for Distribution, Rx for Processing 10/10/2014 Real-Time Innovations, Inc. 17 DR DR DR Observable Observer DW DW Processing

DDS and Rx: A Great Match 10/10/2014 © 2012 RTI • COMPANY 19 DDS Concept Rx Concept/Type/Operator Topic of type T An objectthat implements IObservable<T>, which internally creates a DataReader<T> Communication status, Discovery event streams IObservable<SampleLostStatus> IObservable<SubscriptionBuiltinTopicData> Topic of type T with key type=Key IObservable<IGroupedObservable<Key, T>> Detect a new instance Notify Observers about a new IGroupedObservable<Key, T>with key==instance. Invoke IObserver<IGroupedObservable<Key, T>>.OnNext() Dispose an instance Notify Observers through IObserver<IGroupedObservable<Key,T>>.OnCompleted() Take an instance update of type T Notify Observers about a new value of T using Iobserver<T>.OnNext() Readwith history=N IObservable<T>.Replay(N)(Produces a new IObservable<T>)

DDS and Rx: A Great Match 10/10/2014 © 2012 RTI • COMPANY 20 DDS Concept Rx Concept/Type/Operation QueryConditions Iobservable<T>.Where(…) OR Iobservable<T>.GroupBy(…) SELECT in CFT expression IObservable<T>.Select(...) FROM in CFT expression DDSObservable.FromTopic(“Topic1”) DDSObservable.FromKeyedTopic(“Topic2”) WHEREin CFT expression IObservable<T>.Where(...) ORDER BY in CFT expression IObservable<T>.OrderBy(...) MultiTopic(INNER JOIN) IObservable<T>.Join(...) .Where(...) .Select(...) Join between DDS and non- DDS data Join, CombineLatest, Zip

Living Demo in C# 10/10/2014 Real-Time Innovations, Inc. 21

Background Code 10/12/2014 Real-Time Innovations, Inc. 22 publicclassShapeType: ICopyable<ShapeType> { publicstringcolor; publicintshapesize; publicintx; publicinty; publicShapeType(); } publicclassShapeTypeExtended: ShapeType{ publicfloatangle; publicShapeFillKindfillKind; publicShapeTypeExtended(); } privateDDS.TypedDataWriter<ShapeTypeExtended> triangle_writer;

Flower Demo 10/12/2014 Real-Time Innovations, Inc. 23 IDisposableflower(DDS.DomainParticipantparticipant) { inta = 30, b = 30, c = 10; returnObservable.Interval(TimeSpan.FromMilliseconds(1), Scheduler.Immediate) .Select((longx) => { intangle = (int)(x % 360); returnnewShapeTypeExtended{ x = (int)(120 + (a + b) * Math.Cos(angle) + b * Math.Cos((a / b -c) * angle)), y = (int)(120 + (a + b) * Math.Sin(angle) + b * Math.Sin((a / b -c) * angle)), color = "GREEN", shapesize= 5}; }) .Subscribe(triangle_writer); }

Simple Square to Triangle Transformation 10/12/2014 Real-Time Innovations, Inc. 24 IDisposableforward_verbose(DDS.DomainParticipantparticipant) { varrx_reader= DDSObservable.FromTopic<ShapeTypeExtended>(participant, "Square"); IDisposabledisposable = rx_reader.OnDataAvailable((ShapeTypeExtendedshape) => { DDS.InstanceHandle_thandle = DDS.InstanceHandle_t.HANDLE_NIL; triangle_writer.write(shape, refhandle); }); returndisposable; } IDisposableforward_shortest(DDS.DomainParticipantparticipant) { returnDDSObservable.FromTopic<ShapeTypeExtended>(participant, "Square") .OnDataAvailable(triangle_writer); }

Swap Square’s x and y and Propagate Stream Dispose Event 10/12/2014 Real-Time Innovations, Inc. 25 IDisposableswap(DDS.DomainParticipantparticipant) { returnDDSObservable.FromTopic<ShapeTypeExtended>(participant, "Square") .Select(shape => newShapeTypeExtended{ x = shape.y, y = shape.x, color = shape.color, shapesize= shape.shapesize}) .SubscribeAndDisposeOnCompleted(triangle_writer, newShapeTypeExtended{ color = "BLUE"}); }

Aggregator (Square + Circle = Triangle!) 10/12/2014 Real-Time Innovations, Inc. 26 IDisposableaggregator(DDS.DomainParticipantparticipant) { varrx_square_reader= DDSObservable.FromTopic<ShapeTypeExtended>(participant, "Square"); varrx_circle_reader= DDSObservable.FromTopic<ShapeTypeExtended>(participant, "Circle"); returnnewCompositeDisposable( newIDisposable[] { rx_square_reader.Subscribe(triangle_writer), rx_circle_reader.Subscribe(triangle_writer) } ); }

Square/Circle Correlatorusing LINQ 10/12/2014 Real-Time Innovations, Inc. 27 IDisposableselectmany_correlator(DDS.DomainParticipantparticipant, booluseLinq) { varrx_circle_reader= DDSObservable.FromTopic<ShapeTypeExtended>(participant, "Circle", Scheduler.Default); varrx_square_reader= DDSObservable.FromTopic<ShapeTypeExtended>(participant, "Square", Scheduler.Default); varcorrelator= fromsquare inrx_square_readerfromcircle inrx_circle_reader.Take(1) wheresquare.color== circle.colorselectnewShapeTypeExtended{ x = square.x, y = square.y, color = square.color, shapesize= circle.x}; returncorrelator.Subscribe(triangle_writer); }

Thank You! •Rx4DDS.NET –https://github.com/rticommunity/rticonnextdds-reactive •Data Distribution Service –http://portals.omg.org/dds •Real-Time Innovations, Inc. –www.rti.com 10/12/2014 Real-Time Innovations, Inc. 28

Reactive Stream Processing Using DDS and Rx

More Related Content

What's hot

Viewers also liked

Similar to Reactive Stream Processing Using DDS and Rx

More from Sumant Tambe

Recently uploaded

Reactive Stream Processing Using DDS and Rx