如何获取网站的seo,烟台高新区规划国土建设局网站,wordpress分享可见,重庆住房城乡建设网站响应式编程五 使用StepVerifier测试响应式流StepVerifier要点 使用StepVerifier进行高级测试股票订阅系统数据库表 使用StepVerifier测试响应式流
出于测试目的#xff0c;Reactor 提供了额外的 reactor-test 模块#xff0c;该模块提供了 StepVerifier。StepVerifier 提供了… 响应式编程五 使用StepVerifier测试响应式流StepVerifier要点 使用StepVerifier进行高级测试股票订阅系统数据库表 使用StepVerifier测试响应式流
出于测试目的Reactor 提供了额外的 reactor-test 模块该模块提供了 StepVerifier。StepVerifier 提供了一个流式 API用于为任何 Publisher 构建验证流程。
StepVerifier要点
验证 Publisher 主要有两种方法。第一种是 StepVerifier.Tcreate(PublisherT source)。使用此技术构建的测试如下所示
StepVerifier.create(Flux.just(foo, bar)).expectSubscription().expectNext(foo).expectNext(bar).expectComplete().verify();在此示例中Publisher 应生成两个特定元素后续操作将验证特定元素是否已传递给最终订阅者
该类提供的构建器技术可以
定义验证过程中事件发生的顺序。
根据前面的代码第一个发出的事件必须是与订阅相关的事件紧跟其后的事件必须是 foo 和bar 字符串。 最后 StepVerifier#expectCompletion 定义终止信号的存在。 在此例中必须是 Subscriber#onComplete 的调用或者成功完成给定的 Flux。 要执行验证或者说对创建流进行订阅就必须调用 .verify() 方法。 verify() 是一个阻塞调用它阻塞执行直到流发出所有预期的事件。 通过使用这种简单的技术可以使用可计数的元素和事件来验证 Publisher。但是用大量元素来验证流程是很困难的。 如果检查的是 该发布者已发出元素是否达到特定数量可以使用 .expectNextCount() 。 StepVerifier// .create(Flux.range(0, 101)).create(Flux.range(0, 100))// .create(Flux.range(0, 99)).expectSubscription().expectNext(0) // 期望下一个元素与指定的相等.expectNextCount(98) // 从上个期望或从开始订阅开始期望收到指定个数的元素.expectNext(99) // 期望下一个元素与指定的相等.expectComplete() // 期望收到onComplete信号.verify(); // 阻塞验证
}尽管 .expectNextCount() 方法解决了一部分问题但在某些情况下仅仅检查发出元素的数量是不够的
例如在
验证负责按特定规则过滤或选择元素的代码时检查所有发出的项是否与过滤规则匹配非常重要。
为此StepVerifier 可以使用 Java Hamcrest 等工具立即记录发出的数据及其验证。
如下代码
PublisherWallet usersWallets findAllUsersWallets();StepVerifier.create(usersWallets).expectSubscription().recordWith(ArrayList::new).expectNextCount(1).consumeRecordedWith(wallets - assertThat(wallets,everyItem(hasProperty(owner, equalTo(admin))))).expectComplete().verify();与前面的示例相反每个期望仅涵盖一个元素或指定数量元素的验证 .consumeRecordedWith()可以验证给定 Publisher 发布的所有元素。应该注意的是 .consumeRecordedWith() 只有在指定了.recordWith() 时才有效。反过来我们应该仔细定义存储记录的集合类。对于多线程发布者而言用于记录事件的集合类型应该支持并发访问因此在这些情况下最好使用
.recordWith(ConcurrentLinkedQueue :: new) 而不是 .recordWith(ArrayList :: new) 因为与 ArrayList 相比ConcurrentLinkedQueue 是线程安全的。
除此之外还有其他功能相似的方法。例如对下一个元素的期望的定义如以下代码所示
StepVerifier.create(Flux.just(alpha-foo, betta-bar)).expectSubscription().expectNextMatches(e - e.startsWith(alpha)).expectNextMatches(e - e.startsWith(betta)).expectComplete().verify();expectNextMatches() 和 .expectNext() 之间的唯一区别是前者可以定义自定义的匹配器Predicate 这使其比后者更灵活。这是因为 .expectNext() 基于元素之间的比较而这种比较使用元素的 .equals() 方法
类似地 .assertNext() 和 .consumeNextWith() 使编写自定义断言成为可能。要注意 .assertNext() 是 .consumeNextWith() 的别名。 .expectNextMatches() 和.assertNext() 之间的区别在于前者接受 Predicate必须返回 true 或 false而后者接收可能抛出异常的Consumer并且捕获消费者抛出的任何 AssertionError然后通过 .verify() 方法抛出。 如下面的代码所示 StepVerifier.create(findUsersUSDWallet()).expectSubscription().assertNext(wallet - assertThat(wallet, hasProperty(currency,equalTo(USD))) //有currency属性并且属性值等于USD).expectComplete().verify();最后只剩下未覆盖的错误情况这也是正常系统生命周期的一部分。可以检查错误信号的API 方法不是很多最简单的是 .expectError() 方法该方法没有参数。 如下代码
StepVerifier.create(Flux.error(new RuntimeException(Error)))
.expectError()
.verify();StepVerifier.create(Flux.error(new RuntimeException(Error)))
.expectError()
.verify();在某些情况下测试特定错误类型至关重要。例如如果用户在登录期间输入了错误的凭据则安全服务应发出 BadCredentialsException.class 。为了验证发出的错误我们可以使用.expectError(Class? extends Throwable)
如以下代码所示
StepVerifier.create(securityService.login(admin, wrong))
.expectSubscription()
.expectError(BadCredentialsException.class)
.verify();使用StepVerifier进行高级测试
发布者测试的第一步是验证无界Publisher。根据响应式流规范无限流意味着流永远不会调用Subscriber#onComplete() 方法。由于 StepVerifier 将无限期地等待完成信号因此测试将被阻塞直到它被杀死。
为了解决这个问题StepVerifier 提供了一个
取消 API在满足某些期望时它可以取消对源的订阅。
如下面的代码所示
FluxString webSocketPublisher ...;
StepVerifier.create(websocketPublisher).expectSubscription().expectNext(Connected).expectNext(Price: $12.00).thenCancel() // 取消订阅.verify();上述代码表示在收到 Connected 以及 Price:$ 12.00 消息后我们将断开或取消订阅WebSocket。
系统验证过程的另一个关键阶段是检查 Publisher 的背压行为。例如通过 WebSocket与外部系统交互会产生一个只推式的 Publisher。防止此类行为的一种简单方法是使用.onBackpressureBuffer() 操作符保护下游。要使用所选的背压策略检查系统是否按预期运行必须手动控制用户需求。为此StepVerifier 提供了 .thenRequest() 方法它允许我们控制用户需求。
这由以下代码描述 FluxString websocketPublisher ...;ClassException expectedErrorClass reactor.core.Exceptions.failWithOverflow().getClass();StepVerifier.create(websocketPublisher.onBackpressureBuffer(5), 0) // 使用背 压控制.expectSubscription().thenRequest(1).expectNext(Connected).thenRequest(1).expectNext(Price: $12.00).expectError(expectedErrorClass)在前面的示例中使用的是 StepVerifier.create() 方法的重载它接收初始订阅者的请求作为第二个参数。在单参数方法的重载中默认需求是 Long.MAX_VALUE 即无限需求。
股票订阅系统
数据库表 SET NAMES utf8mb4;
SET FOREIGN_KEY_CHECKS 0;-- ----------------------------
-- Table structure for stock_do
-- ----------------------------
DROP TABLE IF EXISTS stock_do;
CREATE TABLE stock_do (symbol varchar(255) CHARACTER SET utf8mb4 NOT NULL,name varchar(255) CHARACTER SET utf8mb4 NULL DEFAULT NULL,PRIMARY KEY (symbol) USING BTREE
) ENGINE InnoDB CHARACTER SET utf8mb4 ROW_FORMAT Dynamic;-- ----------------------------
-- Records of stock_do
-- ----------------------------
INSERT INTO stock_do VALUES (AMD, AMD Yes!);
INSERT INTO stock_do VALUES (AMZN, 亚马逊);
INSERT INTO stock_do VALUES (APPL, 苹果);
INSERT INTO stock_do VALUES (BABA, 阿里巴巴);
INSERT INTO stock_do VALUES (GOOG, 谷歌);
INSERT INTO stock_do VALUES (INTC, 英特尔);
INSERT INTO stock_do VALUES (KO, 可口可乐);
INSERT INTO stock_do VALUES (MSFT, 微软);
INSERT INTO stock_do VALUES (NFLX, 奈飞);
INSERT INTO stock_do VALUES (PYPL, 贝宝);
INSERT INTO stock_do VALUES (SBUX, 星巴克);
INSERT INTO stock_do VALUES (TSLA, 特斯拉);
INSERT INTO stock_do VALUES (XIACY, 小米);-- ----------------------------
-- Table structure for stock_subscription_do
-- ----------------------------
DROP TABLE IF EXISTS stock_subscription_do;
CREATE TABLE stock_subscription_do (id int(0) NOT NULL AUTO_INCREMENT,email varchar(255) CHARACTER SET utf8mb4 NULL DEFAULT NULL,symbol varchar(255) CHARACTER SET utf8mb4 NULL DEFAULT NULL,PRIMARY KEY (id) USING BTREE
) ENGINE InnoDB AUTO_INCREMENT 10003 CHARACTER SET utf8mb4 ROW_FORMAT Dynamic;-- ----------------------------
-- Records of stock_subscription_do
-- ----------------------------
INSERT INTO stock_subscription_do VALUES (10001, lsdqq.com, TSLA);
INSERT INTO stock_subscription_do VALUES (10002, lsdqq.com, AMZN);
INSERT INTO stock_subscription_do VALUES (10003, lsdqq.com, APPL);
INSERT INTO stock_subscription_do VALUES (10004, lsdqq.com, XIACY);SET FOREIGN_KEY_CHECKS 1;
引入依赖 dependenciesdependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-thymeleaf/artifactId/dependency!--对参数的校验--dependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-validation/artifactId/dependencydependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-data-r2dbc/artifactId/dependencydependencygroupIddev.miku/groupIdartifactIdr2dbc-mysql/artifactIdscoperuntime/scope/dependencydependencygroupIdmysql/groupIdartifactIdmysql-connector-java/artifactIdscoperuntime/scope/dependencydependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-webflux/artifactId/dependencydependencygroupIdorg.projectlombok/groupIdartifactIdlombok/artifactId/dependency!--和前端相关--!--帮我们实现了定义好的表格或者按钮之类--dependencygroupIdorg.webjars/groupIdartifactIdbootstrap/artifactIdversion4.6.0-1/version/dependency!--这里方便我们用javaScript书写一些逻辑--dependencygroupIdorg.webjars/groupIdartifactIdwebjars-locator/artifactIdversion0.30/version/dependencydependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter/artifactId/dependencydependencygroupIdorg.springframework.boot/groupIdartifactIdspring-boot-starter-test/artifactIdscopetest/scopeexclusionsexclusiongroupIdorg.junit.vintage/groupIdartifactIdjunit-vintage-engine/artifactId/exclusion/exclusions/dependency/dependencies
yml配置数据库连接信息 创建如上目录文件。
package com.lsd.dao;import com.lsd.entry.StockDO;
import org.springframework.data.repository.reactive.ReactiveCrudRepository;
import org.springframework.stereotype.Repository;Repository
public interface StockDao extends ReactiveCrudRepositoryStockDO,String {
}
package com.lsd.dao;import com.lsd.entry.StockSubscriptionDO;
import org.springframework.data.repository.reactive.ReactiveCrudRepository;
import org.springframework.stereotype.Repository;
import reactor.core.publisher.Flux;Repository
public interface StockSubscriptionDao extends ReactiveCrudRepositoryStockSubscriptionDO,String {
//findByEmail(String email)是一个自定义的查询方法。Spring Data能够根据方法名来自动推断执行的查询。在这个例子中Spring Data会生成一个查询该查询基于StockSubscriptionDO实体的email属性来查找匹配的记录。FluxStockSubscriptionDO findByEmail(String email);
}
package com.lsd.entry;import lombok.Data;
import lombok.NoArgsConstructor;
import org.springframework.data.annotation.Id;Data
NoArgsConstructor
public class StockDO {Idprivate Long id;private String symbol;private String name;
}
package com.lsd.entry;import lombok.Data;
import lombok.NoArgsConstructor;
import org.springframework.data.annotation.Id;Data
NoArgsConstructor
public class StockSubscriptionDO {Idprivate Long id;private String email;private String symbol;
}
创建Controller及service
package com.lsd.webflux;public final class Constants {private Constants(){}public static final String TEST_USER_EMAIL lsdqq.com;}
package com.lsd.webflux.controller;import com.lsd.webflux.Constants;
import com.lsd.webflux.dto.StockPrice;
import com.lsd.webflux.service.StockPriceService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Controller;
import org.springframework.ui.Model;
import org.springframework.web.bind.annotation.RequestMapping;
import org.thymeleaf.spring5.context.webflux.ReactiveDataDriverContextVariable;
import reactor.core.publisher.Flux;Controller
public class WebController {Autowiredprivate StockPriceService stockPriceService;RequestMapping(/)public String index(Model model){FluxStockPrice price stockPriceService.getPrice(Constants.TEST_USER_EMAIL);model.addAttribute(email,Constants.TEST_USER_EMAIL);model.addAttribute(stockPrices,new ReactiveDataDriverContextVariable(price));return index;}}
package com.lsd.webflux.service;import com.lsd.webflux.dto.StockPrice;
import reactor.core.publisher.Flux;public interface StockPriceService {FluxStockPrice getPrice(String testUserEmail);
}
package com.lsd.webflux.service;import com.lsd.dao.StockSubscriptionDao;
import com.lsd.entry.StockSubscriptionDO;
import com.lsd.webflux.dto.StockPrice;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Flux;
Slf4j
Service
public class StockPriceServiceImpl implements StockPriceService{Autowiredprivate StockSubscriptionDao stockSubscriptionDao;Autowiredprivate PriceQueryEngine priceQueryEngine;Overridepublic FluxStockPrice getPrice(String email) {return doGetPrice(email);}public FluxStockPrice doGetPrice(String email){log.info(获取股票价格信息:{},email);FluxStockSubscriptionDO subscriptions stockSubscriptionDao.findByEmail(email);return subscriptions.map(stockSubscriptionDO - stockSubscriptionDO.getSymbol()).map(symbol - new StockPrice(symbol,priceQueryEngine.getPriceForSymbol(symbol)));}
}
package com.lsd.webflux.dto;import lombok.AllArgsConstructor;
import lombok.Data;Data
AllArgsConstructor
public class StockPrice {private String stock;private String price;
}
package com.lsd.webflux.service;import org.springframework.stereotype.Component;import java.util.Random;Component
public class PriceQueryEngine {//让股票价格随时变化public String getPriceForSymbol(String symbol){Random random new Random();double value random.nextDouble() * 300 100;return String.format(%.2f,value);}
}
处理订阅信息
package com.lsd.webflux.controller;import com.lsd.webflux.Constants;
import com.lsd.webflux.model.StockSubsciption;
import com.lsd.webflux.model.StockSymbol;
import com.lsd.webflux.service.SubscriptionsService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Controller;
import org.springframework.ui.Model;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.ModelAttribute;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.thymeleaf.spring5.context.webflux.ReactiveDataDriverContextVariable;
import reactor.core.publisher.Flux;Controller
RequestMapping(/subscriptions)
public class SubScriptionController {Autowiredprivate SubscriptionsService subscriptionsService;PostMappingpublic String addSubscription(ModelAttribute(value stockSymbol)StockSymbol symbol){String email Constants.TEST_USER_EMAIL;subscriptionsService.addSubscription(email,symbol.getSymbol()).subscribe();return redirect:/subscriptions?added symbol.getSymbol();}GetMappingpublic String subscription(Model model){// 获取订阅的股票FluxStockSubsciption subscriptions subscriptionsService.findByEmail(Constants.TEST_USER_EMAIL);model.addAttribute(email,Constants.TEST_USER_EMAIL);model.addAttribute(subscriptions,new ReactiveDataDriverContextVariable(subscriptions));return subscription;}}
package com.lsd.webflux.service;import com.lsd.entry.StockSubscriptionDO;
import com.lsd.webflux.model.StockSubsciption;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;public interface SubscriptionsService {FluxStockSubsciption findByEmail(String testUserEmail);MonoStockSubscriptionDO addSubscription(String email, String symbol);
}
package com.lsd.webflux.service;import com.lsd.dao.StockSubscriptionDao;
import com.lsd.entry.StockSubscriptionDO;
import com.lsd.webflux.model.StockSubsciption;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;Slf4j
Service
public class SubscriptionsServiceImpl implements SubscriptionsService{Autowiredprivate StockSubscriptionDao subscriptionDao;Overridepublic FluxStockSubsciption findByEmail(String email) {return doFindByEmail(email);}Overridepublic MonoStockSubscriptionDO addSubscription(String email, String symbol) {return doAddSubscprition(email,symbol);}private MonoStockSubscriptionDO doAddSubscprition(String email, String symbol) {log.info(增加订阅信息{},symbol);StockSubscriptionDO subscriptionDO new StockSubscriptionDO();subscriptionDO.setEmail(email);subscriptionDO.setSymbol(symbol);MonoStockSubscriptionDO save subscriptionDao.save(subscriptionDO);return save;}private FluxStockSubsciption doFindByEmail(String email) {log.info(通过邮件地址获取订阅股票信息{},email);return subscriptionDao.findByEmail(email).map(stockSubscriptionDO -StockSubsciption.builder().symbol(stockSubscriptionDO.getSymbol()).email(stockSubscriptionDO.getEmail()).build());}
}
新增订阅获取所有股票信息
package com.lsd.webflux.controller;import com.lsd.webflux.model.Stock;
import com.lsd.webflux.model.StockSymbol;
import com.lsd.webflux.service.StockService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Controller;
import org.springframework.ui.Model;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.thymeleaf.spring5.context.webflux.ReactiveDataDriverContextVariable;
import reactor.core.publisher.Flux;Controller
RequestMapping(/stocks)
public class StockController {Autowiredprivate StockService stockService;GetMappingprivate String getStocks(Model model){FluxStock stocks stockService.getAllStocks();model.addAttribute(stocks,new ReactiveDataDriverContextVariable(stocks));model.addAttribute(stockSymbol,new StockSymbol());return stocks;}
}
package com.lsd.webflux.service;import com.lsd.webflux.model.Stock;
import reactor.core.publisher.Flux;public interface StockService {FluxStock getAllStocks();
}
package com.lsd.webflux.service;import com.lsd.dao.StockDao;
import com.lsd.webflux.model.Stock;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Flux;Slf4j
Service
public class StockServiceImpl implements StockService{Autowiredprivate StockDao stockDao;Overridepublic FluxStock getAllStocks() {return doGetAllStocks();}private FluxStock doGetAllStocks() {log.info(获取所有的股票信息);return stockDao.findAll().map(stockDO - Stock.builder().symbol(stockDO.getSymbol()).name(stockDO.getName()).build());}
}
