package com.yeshimin.ysmspace.config;

import org.elasticsearch.client.RestHighLevelClient;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.elasticsearch.client.ClientConfiguration;
import org.springframework.data.elasticsearch.client.RestClients;
import org.springframework.data.elasticsearch.config.AbstractElasticsearchConfiguration;

@Configuration
public class ElasticsearchConfiguration extends AbstractElasticsearchConfiguration {

    @Override
    public RestHighLevelClient elasticsearchClient() {
        final ClientConfiguration clientConfiguration = ClientConfiguration.builder()
                .connectedTo("vhost4:9200") // 指定es服务地址
                .build();

        return RestClients.create(clientConfiguration).rest();
    }
}

package com.yeshimin.ysmspace.model.elasticsearch;

// ... import

@Data
@Document(indexName = Common.PROJECT_NAME + "_product")
public class ProductDoc {

    /**
     * 主键ID
     */
    @Id
    @Field(name = "id", type = FieldType.Long)
    private Long id;

    /**
     * 创建时间
     */
    @Field(name = "create_time", type = FieldType.Date, format = DateFormat.basic_date_time)
    private LocalDateTime createTime;

    /**
     * 商品名称
     */
    @Field(name = "name", type = FieldType.Keyword)
    private String name;

    // ...
}

package com.yeshimin.ysmspace.dal.elasticsearch;

// ... import

@Repository
public interface ProductEsRepository extends ElasticsearchRepository<ProductDoc, Long> {
}

@Autowired
private ProductEsRepository productEsRepository;

ProductDoc productDoc = new ProductDoc();
// ...
productEsRepository.save(productDoc);

package com.yeshimin.ysmspace.config;

import org.springframework.amqp.core.AmqpAdmin;
import org.springframework.amqp.core.Binding;
import org.springframework.amqp.core.DirectExchange;
import org.springframework.amqp.core.Queue;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitAdmin;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.HashMap;
import java.util.Map;

@Configuration
public class RabbitMqConfiguration {

    public static final String EXCHANGE_DEFAULT_DIRECT = "amq.direct";
    public static final String QUEUE_ORDER_PENDING = "q.order.pending";
    public static final String ROUTING_ORDER_PENDING = "rk.order.pending";
    public static final String DLX_ORDER_PENDING = EXCHANGE_DEFAULT_DIRECT;
    public static final String DLQ_ORDER_PENDING = "q.dl.order.pending";
    public static final String DLR_ORDER_PENDING = "rk.dl.order.pending";

    @Bean
    public AmqpAdmin amqpAdmin(ConnectionFactory connectionFactory) {
        return new RabbitAdmin(connectionFactory);
    }

    @Bean
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        return new RabbitTemplate(connectionFactory);
    }

    @Bean
    public DirectExchange defaultDirectExchange() {
        return new DirectExchange(EXCHANGE_DEFAULT_DIRECT);
    }

    @Bean
    public Queue orderPendingQueue() {
        Map<String, Object> args = new HashMap<>();
//        args.put("x-message-ttl", 30 * 60 * 1000); ttl由消息实体指定
        args.put("x-dead-letter-exchange", DLX_ORDER_PENDING);
        args.put("x-dead-letter-routing-key", DLR_ORDER_PENDING);
        return new Queue(QUEUE_ORDER_PENDING, true, false, false, args);
    }

    @Bean
    public Queue orderPendingDeadLetterQueue() {
        return new Queue(DLQ_ORDER_PENDING);
    }

    @Bean
    public Binding orderPendingBinding() {
        return new Binding(QUEUE_ORDER_PENDING, Binding.DestinationType.QUEUE,
                EXCHANGE_DEFAULT_DIRECT, ROUTING_ORDER_PENDING, null);
    }

    @Bean
    public Binding orderPendingDeadLetterBinding() {
        return new Binding(DLQ_ORDER_PENDING, Binding.DestinationType.QUEUE,
                DLX_ORDER_PENDING, DLR_ORDER_PENDING, null);
    }
}

package com.yeshimin.ysmspace.rabbitmq;

import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessagePostProcessor;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

@Component
public class DefaultMessagePostProcessor implements MessagePostProcessor {

    @Value("${order.timeout}")
    private Integer orderTimeout;

    @Override
    public Message postProcessMessage(Message message) throws AmqpException {
        message.getMessageProperties().setExpiration(String.valueOf(orderTimeout));
        return message;
    }
}

@Autowired
private RabbitTemplate rabbitTemplate;
@Autowired
private DefaultMessagePostProcessor defaultMessagePostProcessor;

// ...

rabbitTemplate.convertAndSend(
    RabbitMqConfiguration.EXCHANGE_DEFAULT_DIRECT,
    RabbitMqConfiguration.ROUTING_SYNC_TO_ES,
    JSON.toJSONString(message), defaultMessagePostProcessor);

// ...

package com.yeshimin.ysmspace.rabbitmq;

// ... import

@Slf4j
@Component
public class DefaultRabbitListener {

    @Autowired
    private OrderAppService orderAppService;

    @RabbitListener(queues = RabbitMqConfiguration.DLQ_ORDER_PENDING)
    public void listen(String message) {
        log.debug("listen(), message: {}", message);

        OrderPendingMessage jsonData = JSON.parseObject(message, OrderPendingMessage.class);

        try {
            orderAppService.cancel(jsonData.getOrderId());
        } catch (Exception e) {
            log.warn("未成功取消订单！");
            e.printStackTrace();
        }
    }
}

@Primary
@Component
public class CustomSwaggerResourcesProvider implements SwaggerResourcesProvider {

    @Lazy
    @Autowired
    private RouteLocator routeLocator;

    @Override
    public List<SwaggerResource> get() {
        List<Route> routes = new ArrayList<>();
        routeLocator.getRoutes().subscribe(routes::add);
        return routes.stream().map(route -> this.resource(route.getUri().getHost())).collect(Collectors.toList());
    }

    private SwaggerResource resource(String service) {
        SwaggerResource swaggerResource = new SwaggerResource();
        swaggerResource.setName(service);
        swaggerResource.setUrl("/" + service + "/v3/api-docs");
        swaggerResource.setSwaggerVersion("3.0.3");
        return swaggerResource;
    }
}

@Order(Ordered.HIGHEST_PRECEDENCE + 100)
public class PrefixedWebMvcOpenApiTransformationFilter extends WebMvcBasePathAndHostnameTransformationFilter {

    private static final String HEADER_X_FORWARDED_PREFIX = "X-Forwarded-Prefix";

    public PrefixedWebMvcOpenApiTransformationFilter(String oasPath) {
        super(oasPath);
    }

    @Override
    public OpenAPI transform(OpenApiTransformationContext<HttpServletRequest> context) {
        OpenAPI openApi = context.getSpecification();
        context.request().ifPresent(servletRequest -> {
            String headerValue = servletRequest.getHeader(HEADER_X_FORWARDED_PREFIX);
            if (!StringUtils.isEmpty(headerValue)) {
                openApi.getServers().forEach(server -> server.setUrl(server.getUrl() + headerValue));
            }
        });
        return openApi;
    }
}

@Configuration
public class SwaggerConfiguration {

    @Bean
    public WebMvcOpenApiTransformationFilter prefixedWebMvcOpenApiTransformer(
            @Value(springfox.documentation.oas.web.SpecGeneration.OPEN_API_SPECIFICATION_PATH) String oasPath) {
        return new PrefixedWebMvcOpenApiTransformationFilter(oasPath);
    }
}

@EnableWebFluxSecurity
public class WebSecurityConfiguration {

    @Autowired
    private JwtTokenAuthenticationManager jwtTokenAuthenticationManager;

    @Bean
    public SecurityWebFilterChain springSecurityFilterChain(ServerHttpSecurity http) {
        // 鉴权配置
        AuthenticationWebFilter authenticationWebFilter = new AuthenticationWebFilter(jwtTokenAuthenticationManager);
        authenticationWebFilter.setServerAuthenticationConverter(new JwtTokenAuthenticationConverter());
        authenticationWebFilter.setSecurityContextRepository(new WebSessionServerSecurityContextRepository());
        http.addFilterAt(authenticationWebFilter, SecurityWebFiltersOrder.AUTHENTICATION);

        // 清除一次请求后的SecurityContext
        http.addFilterBefore(new ClearExchangeSecurityContextWebFilter(), SecurityWebFiltersOrder.REACTOR_CONTEXT);

        // 跨域配置
        http.cors().configurationSource(corsConfigurationSource());

        http.csrf().disable();

        http.authorizeExchange()
                .pathMatchers("/**/login").permitAll()
                .pathMatchers("/**/logout").authenticated()
                .pathMatchers("/upms/**").authenticated()
                .pathMatchers("/product/**").authenticated()
                .anyExchange().denyAll();

        return http.build();
    }

    @Bean
    public CorsConfigurationSource corsConfigurationSource() {
        UrlBasedCorsConfigurationSource source = new UrlBasedCorsConfigurationSource();
        CorsConfiguration config = new CorsConfiguration();

        config.addAllowedHeader("*");
        config.addAllowedMethod("*");
        config.addAllowedOrigin("*");
        config.setAllowCredentials(true);
        config.setMaxAge(3600L);

        source.registerCorsConfiguration("/**", config);
        return source;
    }
}

public class JwtTokenAuthenticationConverter implements ServerAuthenticationConverter {

    @Override
    public Mono<Authentication> convert(ServerWebExchange exchange) {
        return apply(exchange);
    }

    public Mono<Authentication> apply(ServerWebExchange exchange) {
        ServerHttpRequest request = exchange.getRequest();

        String token = request.getHeaders().getFirst(HeaderNames.X_TOKEN);
        if (StringUtils.isBlank(token)) {
            return Mono.empty();
        }

        return Mono.just(new JwtTokenAuthenticationToken(token));
    }
}

public class JwtTokenAuthenticationToken extends AbstractAuthenticationToken {

    private String token;
    private Long userId;

    public JwtTokenAuthenticationToken(Collection<? extends GrantedAuthority> authorities) {
        super(authorities);
    }

    public JwtTokenAuthenticationToken(String token) {
        super(null);
        this.token = token;
    }

    public void setUserId(Long userId) {
        this.userId = userId;
    }

    public Long getUserId() {
        return this.userId;
    }

    @Override
    public Object getCredentials() {
        return token;
    }

    @Override
    public Object getPrincipal() {
        return null;
    }

    @Override
    public boolean implies(Subject subject) {
        return false;
    }

@Slf4j
@Component
public class JwtTokenAuthenticationManager implements ReactiveAuthenticationManager {

    @Autowired
    private AuthFeignService authFeignService;

    @Override
    public Mono<Authentication> authenticate(Authentication authentication) throws AuthenticationException {
        JwtTokenAuthenticationToken requestToken = (JwtTokenAuthenticationToken) authentication;
        String token = (String) requestToken.getCredentials();

        AuthDto authDto = new AuthDto();
        authDto.setToken(token);
        // 服务级别只认证，不做授权逻辑判断
        authDto.setOnlyAuthenticate(true);
        ResultVo<AuthVo> resultVo = authFeignService.auth(authDto);
        if (resultVo.getCode() != ErrorCodeEnum.SUCCESS.getCode() || !resultVo.getData().getAuthenticated()) {
            log.error("调用auth服务鉴权失败！");
            return Mono.error(new BadCredentialsException("调用auth服务鉴权失败！"));
        }

        // set authorities
        List<GrantedAuthority> authorities = Collections.emptyList();
        JwtTokenAuthenticationToken resultToken = new JwtTokenAuthenticationToken(authorities);
        resultToken.setAuthenticated(true);
        resultToken.setUserId(resultVo.getData().getUserId());

        return Mono.just(resultToken);
    }
}

public class ClearExchangeSecurityContextWebFilter implements WebFilter {

    private ServerSecurityContextRepository securityContextRepository = new WebSessionServerSecurityContextRepository();

    @Override
    public Mono<Void> filter(ServerWebExchange exchange, WebFilterChain chain) {
        return chain.filter(exchange).then(securityContextRepository.save(exchange, null));
    }
}

public class GatewayWebContextGlobalFilter implements GlobalFilter {

    @Override
    public Mono<Void> filter(ServerWebExchange exchange, GatewayFilterChain chain) {
        return ReactiveSecurityContextHolder.getContext()
                .map(SecurityContext::getAuthentication)
                .filter(authentication -> authentication instanceof JwtTokenAuthenticationToken)
                .map(authentication -> (JwtTokenAuthenticationToken) authentication)
                .flatMap(authentication -> {
                    String headerName = HeaderNames.X_USER_ID;
                    String headerValue = String.valueOf(authentication.getUserId());

                    ServerHttpRequest request = exchange.getRequest().mutate().header(headerName, headerValue).build();

                    return chain.filter(exchange.mutate().request(request).build());
                })
                .switchIfEmpty(chain.filter(exchange));
    }
}

@Configuration
public class GatewayConfiguration {

    @Bean
    public GatewayWebContextGlobalFilter gatewayWebContextGlobalFilter() {
        return new GatewayWebContextGlobalFilter();
    }
}

@EnableGlobalMethodSecurity(prePostEnabled = true)
@EnableWebSecurity
public class WebSecurityConfiguration extends WebSecurityConfigurerAdapter {

    @Autowired
    private JwtTokenAuthenticationProvider jwtTokenAuthenticationProvider;

    @Override
    protected void configure(HttpSecurity http) throws Exception {
        http.csrf().disable();
        http.cors().disable();

        http.addFilterAfter(new JwtTokenAuthenticationFilter(authenticationManagerBean()), LogoutFilter.class);
    }

    /**
     * 如果重写了authenticationManagerBean()，需要同时重写该方法
     *
     * @see WebSecurityConfigurerAdapter#authenticationManagerBean()
     */
    @Override
    protected void configure(AuthenticationManagerBuilder auth) {
        auth.authenticationProvider(jwtTokenAuthenticationProvider);
    }

    @Bean
    @Override
    public AuthenticationManager authenticationManagerBean() throws Exception {
        return super.authenticationManagerBean();
    }
}

public class JwtTokenAuthenticationToken extends AbstractAuthenticationToken {

    private String token;

    public JwtTokenAuthenticationToken(Collection<? extends GrantedAuthority> authorities) {
        super(authorities);
    }

    public JwtTokenAuthenticationToken(String token) {
        super(null);
        this.token = token;
    }

    @Override
    public Object getCredentials() {
        return token;
    }

    @Override
    public Object getPrincipal() {
        return null;
    }

    @Override
    public boolean implies(Subject subject) {
        return false;
    }
}

@Slf4j
public class JwtTokenAuthenticationFilter extends OncePerRequestFilter {

    private AuthenticationManager authenticationManager;

    public JwtTokenAuthenticationFilter(AuthenticationManager authenticationManager) {
        this.authenticationManager = authenticationManager;
    }

    @Override
    protected void doFilterInternal(HttpServletRequest request, HttpServletResponse response, FilterChain filterChain)
            throws ServletException, IOException {

        logger.debug("doFilterInternal()");

        // If required, authenticate and set web context
        authenticateIfRequired(request);

        filterChain.doFilter(request, response);

        // Clear web context
        WebContextUtils.clear();
    }

    /**
     * Authenticate if required
     */
    private void authenticateIfRequired(HttpServletRequest request) {
        String token = request.getHeader(Common.TOKEN_HEADER_KEY);
        if (StringUtils.isBlank(token)) {
            log.debug("no token, as anonymous in later AnonymousAuthenticationFilter");
//            // 在过滤链AnonymousAuthenticationFilter中设置为匿名用户
//            // 这里要先clear，否则在匿名Filter中会存在JwtTokenAuthenticationToken；具体原因暂时未知
            SecurityContextHolder.clearContext();
            return;
        }

        // wrap request authentication
        Authentication authRequest = new JwtTokenAuthenticationToken(token);
        // authenticate for result (delegate by ProviderManager)
        Authentication authResult = authenticationManager.authenticate(authRequest);
        // set to security context
        SecurityContextHolder.getContext().setAuthentication(authResult);
    }
}

@Slf4j
@Component
public class JwtTokenAuthenticationProvider implements AuthenticationProvider {

    @Autowired
    private AuthFeignService authFeignService;

    @Override
    public Authentication authenticate(Authentication authentication) throws AuthenticationException {
        JwtTokenAuthenticationToken requestToken = (JwtTokenAuthenticationToken) authentication;
        String token = (String) requestToken.getCredentials();

        AuthDto authDto = new AuthDto();
        authDto.setToken(token);
        ResultVo<AuthVo> resultVo = authFeignService.auth(authDto);
        if (resultVo.getCode() != ErrorCodeEnum.SUCCESS.getCode() || !resultVo.getData().getAuthenticated()) {
            log.error("调用auth服务鉴权失败！");
            throw new BaseException(ErrorCodeEnum.AUTH_FAIL, "调用auth服务鉴权失败！");
        }

        // set web context
        WebContextUtils.setToken(token);
        WebContextUtils.setUserId(resultVo.getData().getUserId());

        AuthVo authVo = resultVo.getData();
        // role
        List<String> listAuthority = authVo.getRoles().stream().map(s -> "ROLE_" + s).collect(Collectors.toList());
        // add resource
        listAuthority.addAll(authVo.getResources());
        String commaSeparatedRoles = String.join(",", listAuthority);

        // set authorities
        List<GrantedAuthority> authorities = AuthorityUtils.commaSeparatedStringToAuthorityList(commaSeparatedRoles);
        // new authentication
        JwtTokenAuthenticationToken resultToken = new JwtTokenAuthenticationToken(authorities);
        resultToken.setAuthenticated(true);

        return resultToken;
    }

    @Override
    public boolean supports(Class<?> authentication) {
        return JwtTokenAuthenticationToken.class.isAssignableFrom(authentication);
    }
}

@Slf4j
public class ServiceWebContextFilter extends OncePerRequestFilter {

    @Override
    protected void doFilterInternal(HttpServletRequest request, HttpServletResponse response, FilterChain filterChain)
            throws ServletException, IOException {

        String token = request.getHeader(HeaderNames.X_TOKEN);
        String userId = request.getHeader(HeaderNames.X_USER_ID);

        log.debug("WebContext.token: {}", token);
        log.debug("WebContext.userId: {}", userId);

        if (token != null) {
            WebContextUtils.setToken(token);
        }
        if (userId != null) {
            WebContextUtils.setUserId(Long.parseLong(userId));
        }

        filterChain.doFilter(request, response);
    }
}

@Configuration
public class ContextConfiguration {

    @Bean
    public FilterRegistrationBean<ServiceWebContextFilter> serviceWebContextFilter() {
        FilterRegistrationBean<ServiceWebContextFilter> registrationBean = new FilterRegistrationBean<>();

        registrationBean.setFilter(new ServiceWebContextFilter());
        registrationBean.setOrder(Ordered.LOWEST_PRECEDENCE);

        return registrationBean;
    }
}

@Bean
public CorsConfigurationSource corsConfigurationSource() {
    UrlBasedCorsConfigurationSource source = new UrlBasedCorsConfigurationSource();
    CorsConfiguration config = new CorsConfiguration();
    config.addAllowedHeader("*");
    config.addAllowedMethod("*");
    config.addAllowedOrigin("*");
    config.setAllowCredentials(true);
    config.setMaxAge(3600L);
    source.registerCorsConfiguration("/**", config);
    return source;
}

@Bean
@ConditionalOnMissingBean
public HttpMessageConverters messageConverters(ObjectProvider<HttpMessageConverter<?>> converters) {
    return new HttpMessageConverters(converters.orderedStream().collect(Collectors.toList()));
}