幾個月前，我們開始重新設計Drools最低級別的可執行模型 ，并使最終用戶可以使用Java 8 API進行訪問。 為了證明這種方法的靈活性，我嘗試將其與反應流集成在一起，尤其是將該流用作Drools的數據源。

為了說明這是如何工作的，我創建了一個簡單的溫度服務器，該服務器提供RxJava Observable ， 每秒發射給定鎮的溫度，并在5秒鐘后終止。 還有第二種工廠方法，該方法允許合并更多的這些Observable，以使一個Observable可以同時發射一個以上城鎮的溫度。

public class TempServer {public static Observable<TempInfo> getFeed(String town) {return Observable.create(subscriber ->Observable.interval(1, TimeUnit.SECONDS).subscribe(i -> {if (i > 5) subscriber.onCompleted();try {subscriber.onNext(TempInfo.fetch(town));} catch (Exception e) {subscriber.onError(e);}}));}public static Observable<TempInfo> getFeeds(String... towns) {return Observable.merge(Arrays.stream(towns).map(TempServer::getFeed).collect(toList()));} }

其中TempInfo.fetch方法僅返回-20至50度之間的隨機溫度

public TempInfo(String town, int temp) {this.town = town;this.temp = temp; }public static TempInfo fetch(String town) {return new TempInfo(town, random.nextInt(70) - 20); }

使用前一篇文章中介紹的Java 8 DSL的改進版本，我定義了以下2條規則：

Variable<TempInfo> temp = any( TempInfo.class ); Variable<Person> person = any( Person.class );Rule r1 = rule("low temp").view(subscribe(temp, "tempFeed"),expr(temp, t -> t.getTemp() < 0),input(person, "persons"),expr(person, temp, (p, t) -> p.getTown().equals(t.getTown()))).then(on(person, temp).execute((p, t) -> System.out.println(p.getName() + " is freezing in " + p.getTown() + " - temp is " + t.getTemp())));Rule r2 = rule("high temp").view(subscribe(temp, "tempFeed"),expr(temp, t -> t.getTemp() > 30),input(person, "persons"),expr(person, temp, (p, t) -> p.getTown().equals(t.getTown()))).then(on(person, temp).execute((p, t) -> System.out.println(p.getName() + " is sweating in " + p.getTown() + " - temp is " + t.getTemp())));

在這里，我使用2種不同的數據源：一個被動的數據源，可以將其視為事實的存儲：

DataStore persons = storeOf(new Person("Mark", 37, "London"),new Person("Edson", 35, "Toronto"),new Person("Mario", 40, "Milano"));

可以綁定到特定的Drools KieSession與

bindDataSource(ksession, "persons", persons);

以及從上面實現的TempServer中獲取的一個響應式

Observable<TempInfo> tempFeed = TempServer.getFeeds( "Milano", "London", "Toronto" );

也可以以類似的方式綁定到相同的KieSession

bindRxObservable( ksession, "tempFeed", tempFeed );

完成此操作后，您可以觸發這兩個規則并獲得如下輸出：

Mark is freezing in London - temp is -9 Edson is sweating in Toronto - temp is 42 Mario is sweating in Milano - temp is 42 Mario is sweating in Milano - temp is 49 Mark is freezing in London - temp is -17 Edson is sweating in Toronto - temp is 40 Edson is sweating in Toronto - temp is 47 Mario is freezing in Milano - temp is -14 Mark is freezing in London - temp is -8 Mark is freezing in London - temp is -17

• 可在此處找到運行此示例的完整測試用例。

翻譯自: https://www.javacodegeeks.com/2015/11/using-a-rective-stream-as-a-data-source-for-drools.html

創作挑戰賽新人創作獎勵來咯，堅持創作打卡瓜分現金大獎

總結

以上是生活随笔為你收集整理的使用反应流作为Drools的数据源的全部內容，希望文章能夠幫你解決所遇到的問題。

如果覺得生活随笔網站內容還不錯，歡迎將生活随笔推薦給好友。