Implement streaming response output based on SpringBoot and Dify

 <dependency>
            <groupId></groupId>
            <artifactId>spring-boot-starter-webflux</artifactId>
 </dependency>

@Configuration
public class WebConfig implements WebMvcConfigurer {

    static final List<String> ORIGIN_LIST = (
            // Local            "http://localhost:8080",
            "http://127.0.0.1:8080",
            "http://localhost:8888",
            "http://127.0.0.1:8888",
            "http://localhost:8803",
            "http://127.0.0.1:8803"
    );

    @Override
    public void addCorsMappings(CorsRegistry registry) {
        // Configure global cross-domain rules        ("/**") // Allow requests for all paths                .allowedOrigins(ORIGIN_LIST.toArray(new String[0])) // Allowed sources                .allowedMethods("GET", "POST", "PUT", "DELETE", "OPTIONS") // Allowed HTTP methods                .allowedHeaders("Content-Type", "Authorization") // Allowed request header                .allowCredentials(true); // Whether to allow sending cookies and other credential information    }
}

@Slf4j
@RestController
@RequestMapping("/api")
@RequiredArgsConstructor
public class DifyController {
    

    @Value("${}")
    private String chatMessages;
    
    private final DifyService difyService;
    
    @GetMapping(value = "/chatMessagesStreaming", produces = "text/event-stream")
    public Flux<StreamResponse> chatMessagesStreaming(HttpServletRequest request,
                                                  @RequestParam(value = "query", required = true) String query,
                                                  @RequestParam(value = "userName", required = true) String userName,
                                                  @RequestParam(value = "conversationId", required = false) String conversationId) throws Exception {
   
	   return (query, conversationId, userName).doOnNext(response -> {
	        ("Flowing Results:" + ());
	        //The workflow_finished node can obtain the complete answer and perform your logical processing	        if (().equals("workflow_finished")) {
	            ("Enter workflow_finished stage");
	            String answer = ().getOutputs().getAnswer();//Complete answer	        }
	         // Message_end end node, perform your logical processing	        if (().equals("message_end")) {
	            ("Enter message_end");
	        }
	
	    });
}

java
@Slf4j
@Service
@RequiredArgsConstructor
public class DifyService {

	@Value("${}")
    private String url;

    @Value("${}")
    private String apiKey;
	/**
      * Streaming call to diify.
      *
      * @param query text
      * @param conversationId id
      * @param userName Username
      * @return Flux response stream
      */
    public Flux<StreamResponse> streamingMessage(String query, String conversationId, String userName) {
        //1. Set the request body        DifyRequestBody body = new DifyRequestBody();
        (new HashMap<>());
        (query);
        ("streaming");
        ("");
        (userName);
        if ((conversationId)) {
            (conversationId);
        }
        //If there is a custom entry parameter, it can be added to the following map        //Map<String, Object> commoninputs = new HashMap<>();
        //("search_type", searchType);
        //(commoninputs);

        //2. Use webclient to send post request        return ()
                .uri(url)
                .headers(httpHeaders -> {
                    (MediaType.APPLICATION_JSON);
                    (apiKey);
                })
                .bodyValue((body))
                .retrieve()
                .bodyToFlux();// Entity conversion                .filter(this::shouldInclude) // Filter out unnecessary data [Add to increase according to demand]                //.map(this::convertToCustomResponseAsync) // Asynchronous conversion [if the return format is customized, it is implemented through asynchronous conversion]                .onErrorResume(throwable -> {
                	("Exception output:"+())
                })
                //.concatWith((createCustomFinalMessage())); // Add a custom final message [added according to demand]    }

    private boolean shouldInclude(StreamResponse streamResponse) {
        // Example: Just the data of the message node and the data of the message_end node        if (().equals("message")
                || ().equals("message_end")) {
            return true;
        }
        return false;
    }

@Data
public class StreamResponse implements Serializable {

    /**
      * Event types in different modes.
      */
    private String event;

    /**
     * agent_thought id.
     */
    private String id;

    /**
      * Task ID.
      */
    private String task_id;

    /**
      * Message unique ID.
      */
    private String message_id;

    /**
      * LLM Returns the text block content.
      */
    private String answer;

    /**
      * Create timestamp.
      */
    private Long created_at;

    /**
      * Session ID.
      */
    private String conversation_id;
    
    private StreamResponseData data;
}

@Data
public class StreamResponseData implements Serializable {
    private String id;
    private String workflow_id;
    private String status;
    private Long created_at;
    private Long finished_at;
    private OutputsData outputs;
}

@Data
public class OutputsData implements Serializable {
    private String answer;
}