the whole chain

// I call this in the main method
rulebreakerChain() // this is a function that returns a flux<message>
            .onErrorContinue((err, object) -> logger.error("error in rulebreakerchain", err))
            .onErrorResume(err -> {
              logger.error("Error in rulebreaker chain", err);
              return Mono.empty();
            })
            .subscribe();

// this is the aforementioned funciton
private static Flux<Message> rulebreakerChain() {
    return client.getEventDispatcher().on(MessageCreateEvent.class)
            .filter(msg -> msg.getMember().isPresent())
            .filter(/* filters out all but three guilds by the id*/)
            .filterWhen(/* filters out messages from channels whose name contains "giveaways" or advertisements */)
            .filterWhen(msg -> /* filters out all mods or above */ )
            .map(RuleBreakProcessor::new) // i'll include the class bellow this func
            .filter(RuleBreakProcessor::process) // calls a funciton inside of the class
            .filterWhen(proc -> /* filters out messages that contain an invite to another server */ )
            .flatMap(RuleBreakProcessor::getMessages); // see bellow
  }
  
  // this is in the RuleBreakProcessor class
  
  public RuleBreakProcessor(MessageCreateEvent e) {
    // stores the message in this object
  }
  
  public boolean process() {
    return something; // this func does some regex on the message stored inside of this object, returns whether it contains any blacklisted words or invites
  }
  
 public Flux<Message> getMessages() {
    return getNotificationText() // function bellow this one
            .flux()
            .flatMap(msg -> {
              String finalMessage = " The user's ID is: ***" + event.getMember().get().getId().asString() + "***";

              return event.getGuild()
                      .flux()
                      .flatMap(Guild::getMembers)
                      .filterWhen(member -> /* checks if user has one of the needed roles from the DB */)
                      .filterWhen(member -> /* filters out all offline and away users */)
                      .filter(mbmr -> !mbmr.isBot())
                      .flatMap(mbmr ->
                              mbmr.getPrivateChannel()
                                      .onErrorResume((err) -> {
                                        PeepoCop.logger.info("Could not DM " + Utility.getUsernameDetail(mbmr) + " about rule break.");
                                        return Mono.empty();
                                      })
                                      .flatMap(chnl ->
                                              chnl.createMessage(spec ->
                                                      spec.setContent(msg.concat(finalMessage))))
                                      .filterWhen(message -> /* filters them based on a setting from a DB */)
                                      .flatMap(message ->  // this is so they recieve a users DM in a separate message so it's easier to copy on mobile
                                              message.getChannel()
                                                      .flatMap(chnl ->
                                                        chnl.createMessage(spec ->
                                                          spec.setContent(event.getMember().get().getId().asString()))
                                                      )
                                      ) 
                      );
            });
  }

  public Mono<String> getNotificationText() {
    return Mono.zip(event.getMessage().getChannel().ofType(TextChannel.class),
            event.getGuild())
            .map(tpl -> /* this just returns a formated message. */ );
  }