James添加收件监听

package org.apache.james.mailetcontainer.impl.matchers;


import org.apache.james.events.Event;
import org.apache.james.events.EventListener;
import org.apache.james.events.Group;
import org.apache.james.mailbox.MessageUid;
import org.apache.james.mailbox.events.MailboxEvents;
import org.apache.james.mailbox.model.MailboxId;
import org.reactivestreams.Publisher;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import reactor.core.publisher.Mono;

public class UIDEventListener implements EventListener.ReactiveGroupEventListener {

    public static class UIDEventListenerGroup extends Group {

    }

    private static final Group GROUP = new UIDEventListenerGroup();

    private static final String HTTP_ENDPOINT = "http://thirdparty.example.com/api/receive";

    public static final Logger LOGGER = LoggerFactory.getLogger(UIDEventListener.class);


    @Override
    public Publisher<Void> reactiveEvent(Event event) {
        if (event instanceof MailboxEvents.Added) {
            MailboxEvents.Added addedEvent = (MailboxEvents.Added) event;
            MailboxId mailboxId = addedEvent.getMailboxId();
            for (MessageUid uid : addedEvent.getUids()) {
                String jsonData = String.format("{\"mailboxId\": \"%s\", \"uid\": \"%s\"}",
                        mailboxId.serialize(), uid.asLong());
                sendHttpRequest(jsonData);
            }
        }
        return Mono.empty();
    }

    @Override
    public ExecutionMode getExecutionMode() {
        // 这里返回异步模式，若需要同步请改为 ExecutionMode.SYNCHRONOUS（根据实际需求）
        return ExecutionMode.ASYNCHRONOUS;
    }

    @Override
    public boolean isHandling(Event event) {
        // 仅处理 MailboxEvents.Added 事件
        return event instanceof MailboxEvents.Added;
    }

    private void sendHttpRequest(String jsonData) {
        LOGGER.info("收件触发: {}", jsonData);
    }

    @Override
    public Group getDefaultGroup() {
        return GROUP;
    }
}

import com.touchsmail.common.domain.AjaxResult;
import io.swagger.annotations.ApiOperation;
import lombok.extern.slf4j.Slf4j;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestBody;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;

@Slf4j
@RestController
@RequestMapping("/api/mail/hook")
public class EmailHookController {

    /**
     * 邮箱服务器接收邮件回调
     */
    @ApiOperation(value = "邮箱服务器接收邮件回调")
    @PostMapping("/receive")
    public AjaxResult receiveEmails(@RequestBody String param) {
        return AjaxResult.success(param);
    }

}

package org.apache.james.mailetcontainer.impl.matchers;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.io.InputStream;
import java.net.URI;
import java.net.http.HttpClient;
import java.net.http.HttpRequest;
import java.net.http.HttpResponse;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Base64;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.stream.Collectors;

import javax.inject.Inject;
import javax.mail.Address;
import javax.mail.BodyPart;
import javax.mail.Message;
import javax.mail.MessagingException;
import javax.mail.Multipart;
import javax.mail.internet.InternetAddress;
import javax.mail.internet.MimeMessage;

import org.apache.commons.lang3.StringUtils;
import org.apache.james.events.Event;
import org.apache.james.events.EventListener;
import org.apache.james.events.Group;
import org.apache.james.mailbox.MailboxManager;
import org.apache.james.mailbox.MailboxSession;
import org.apache.james.mailbox.MessageManager;
import org.apache.james.mailbox.MessageUid;
import org.apache.james.mailbox.events.MailboxEvents;
import org.apache.james.mailbox.exception.MailboxException;
import org.apache.james.mailbox.model.FetchGroup;
import org.apache.james.mailbox.model.MailboxId;
import org.apache.james.mailbox.model.MessageRange;
import org.reactivestreams.Publisher;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.fasterxml.jackson.databind.ObjectMapper;

import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;


public class UIDEventListener implements EventListener.ReactiveGroupEventListener {

    private final MailboxManager mailboxManager;
    private static final Logger LOGGER = LoggerFactory.getLogger(UIDEventListener.class);

    public static volatile Boolean receiptCallback = false;

    @Inject
    public UIDEventListener(MailboxManager mailboxManager) {
        this.mailboxManager = mailboxManager;
    }

    @Override
    public Publisher<Void> reactiveEvent(Event event) {

        if (!receiptCallback) {
            LOGGER.info("收件回调未打开");
            return Mono.empty();
        } else if (event instanceof MailboxEvents.Added) {
            MailboxEvents.Added addedEvent = (MailboxEvents.Added) event;
            MailboxSession session = mailboxManager.createSystemSession(event.getUsername());

            return Flux.fromIterable(addedEvent.getUids())
                    .flatMap(uid -> processMessage(addedEvent.getMailboxId(), uid, session))
                    .onErrorResume(e -> {
                        LOGGER.error("Message processing failed", e);
                        return Mono.empty();
                    })
                    .then();
        }
        return Mono.empty();
    }

    // 在消息处理链中传递文件夹名称
    private Mono<Void> processMessage(MailboxId mailboxId, MessageUid uid, MailboxSession session) {
        return Mono.usingWhen(
                        Mono.fromCallable(() -> mailboxManager.getMailbox(mailboxId, session)),
                        messageManager -> {
                            String folderName = extractFolderName(messageManager, mailboxId);
                            return fetchMessageContent(messageManager, uid, session)
                                    .flatMap(this::parseMimeMessage)
                                    .flatMap(mimeMessage -> sendNotification(mailboxId, uid, mimeMessage, folderName));
                        },
                        messageManager -> Mono.empty()
                )
                .doOnError(e -> LOGGER.warn("Failed to process UID:{} in mailbox:{}",
                        uid.asLong(), mailboxId.serialize(), e))
                .onErrorResume(e -> Mono.empty());
    }

    // 增加文件夹名称参数传递
    private Mono<Void> sendNotification(MailboxId mailboxId, MessageUid uid,
                                        MimeMessage mimeMessage, String folderName) {
        try {
            sendAsyncHttpRequest(mimeMessage, mailboxId, uid, folderName);
            return Mono.empty();
        } catch (Exception e) {
            return Mono.error(e);
        }
    }


    // 新增文件夹名称提取方法
    private String extractFolderName(MessageManager messageManager, MailboxId mailboxId) {
        try {
            return messageManager.getMailboxPath().getName();
        } catch (Exception e) {
            LOGGER.warn("Cannot get folder name for mailboxId:{}", mailboxId);
            return "unknown_folder";
        }
    }

    // 增加JSON特殊字符转义处理‌:ml-citation{ref="1" data="citationList"}
    private String escapeJsonString(String input) {
        return input.replace("\"", "\\\"")
                .replace("\\", "\\\\")
                .replace("\b", "\\b")
                .replace("\f", "\\f")
                .replace("\n", "\\n")
                .replace("\r", "\\r")
                .replace("\t", "\\t");
    }


    private Mono<byte[]> fetchMessageContent(MessageManager manager, MessageUid uid, MailboxSession session) {
        return Mono.from(manager.getMessagesReactive(MessageRange.one(uid),
                        FetchGroup.FULL_CONTENT, session))
                .publishOn(Schedulers.boundedElastic())
                .handle((messageResult, sink) -> {
                    try (InputStream is = messageResult.getFullContent().getInputStream()) {
                        sink.next(is.readAllBytes());
                    } catch (MailboxException | IOException e) {
                        sink.error(new RuntimeException(e));
                    }
                });
    }

    private Mono<MimeMessage> parseMimeMessage(byte[] content) {
        return Mono.fromCallable(() ->
                new MimeMessage(null, new ByteArrayInputStream(content)));
    }

    private void sendAsyncHttpRequest(MimeMessage message, MailboxId mailboxId, MessageUid uid, String folderName) throws MessagingException, IOException {
        HttpClient httpClient = HttpClient.newHttpClient();
        Map<String, Object> stringObjectMap = buildRequestBody(message, mailboxId, uid, folderName);
        ObjectMapper objectMapper = new ObjectMapper();
        String requestBody = objectMapper.writeValueAsString(stringObjectMap);
        HttpRequest request = HttpRequest.newBuilder()
                .uri(URI.create("http://127.0.0.1:6001/api/mail/hook/receive"))
                .header("Content-Type", "application/json")
                .POST(HttpRequest.BodyPublishers.ofString(requestBody))
                .build();
        // 发送异步请求
        CompletableFuture<HttpResponse<String>> futureResponse = httpClient.sendAsync(request, HttpResponse.BodyHandlers.ofString());
        // 处理异步响应
        futureResponse.thenAccept(response -> {
            int statusCode = response.statusCode();
            String responseBody = response.body();

            if (statusCode == 200) {
                LOGGER.info("收件钩子：推送成功");
            } else {
                LOGGER.error("收件钩子：邮件推送失败，状态码: {} ,响应内容: {} ", statusCode, responseBody);
            }

        }).exceptionally(ex -> {
            LOGGER.error("收件钩子：邮件推送时发生异常: {}", ex.toString());
            return null;
        });
    }

    private Map<String, Object> buildRequestBody(MimeMessage message, MailboxId mailboxId, MessageUid uid, String folderName) throws MessagingException, IOException {
        Map<String, Object> emailData = new HashMap<>();
        // 异步去获取邮件正文
        CompletableFuture<Object> futureResult = handleRelatedContentAsync(message);

        Address[] tos = message.getRecipients(Message.RecipientType.TO);

        Address[] ccs = message.getRecipients(Message.RecipientType.CC);

        Address[] bccs = message.getRecipients(Message.RecipientType.BCC);

        List<String> to = new ArrayList<>();
        List<String> cc = new ArrayList<>();
        List<String> bcc = new ArrayList<>();

        if (tos != null && tos.length > 0) {
            to = Arrays.stream(tos).map(m -> ((InternetAddress) m).getAddress()).collect(Collectors.toList());
        }

        if (ccs != null && ccs.length > 0) {
            cc = Arrays.stream(ccs).map(m -> ((InternetAddress) m).getAddress()).collect(Collectors.toList());
        }

        if (bccs != null && bccs.length > 0) {
            bcc = Arrays.stream(bccs).map(m -> ((InternetAddress) m).getAddress()).collect(Collectors.toList());
        }


        List<String> from = Arrays.stream(message.getFrom()).map(m -> ((InternetAddress) m).getAddress()).collect(Collectors.toList());

        emailData.put("subject", message.getSubject());
        emailData.put("from", from);
        emailData.put("to", to);
        emailData.put("cc", cc);
        emailData.put("bcc", bcc);

        emailData.put("mailboxId", mailboxId.serialize());
        emailData.put("folderName", escapeJsonString(folderName));
        emailData.put("uid", uid.asLong());

        emailData.put("messageID", message.getMessageID());

        emailData.put("date", message.getReceivedDate());
        emailData.put("content", futureResult.join());
        return emailData;
    }

    private CompletableFuture<Object> handleRelatedContentAsync(Message message) {
        return CompletableFuture.supplyAsync(() -> {
            try {
                if (message.isMimeType("text/plain") || message.isMimeType("text/html")) {
                    return message.getContent().toString();
                } else if (message.isMimeType("multipart/*")) {
                    Multipart multipart = (Multipart) message.getContent();
                    StringBuilder htmlContent = new StringBuilder();
                    Map<String, String> imageMap = new HashMap<>();
                    buildMultipartResout(multipart, htmlContent, imageMap);
                    return htmlContent.toString();
                }
                return "unknown content format";
            } catch (Exception e) {
                // 记录错误日志
                LOGGER.error("Error processing related content", e);
                return "error";
            }
        });
    }

    private void buildMultipartResout(Multipart multipart, StringBuilder htmlContent, Map<String, String> imageMap) throws Exception {

        // 遍历所有部分，收集HTML内容和图片
        for (int i = 0; i < multipart.getCount(); i++) {
            BodyPart part = multipart.getBodyPart(i);
            // 处理 multipart/related 类型
            if (part.getContentType().contains("multipart/related") || part.getContentType().contains("multipart/RELATED")) {
                buildMultipartResout((Multipart) part.getContent(), htmlContent, imageMap);
            } else if (part.isMimeType("multipart/alternative") || part.getContentType().contains("multipart/ALTERNATIVE")) {
                processMultipartAlternative(part, htmlContent, imageMap);
            } else if (part.isMimeType("image/*")) {
                processImg(part, imageMap);
            } else if (part.getContent() instanceof Multipart) {
                processMultipartAlternative(part, htmlContent, imageMap);
            } else if (part.isMimeType("text/html")) {
                htmlContent.append(part.getContent().toString());
            } else {
                // 可能是附件
                LOGGER.warn("其他部分 ContentType: " + part.getContentType());
            }
        }

        String content = getContent(htmlContent, imageMap);
        htmlContent.setLength(0);
        htmlContent.append(StringUtils.isEmpty(content) ? "unknown content format" : content);
    }

    private void processMultipartAlternative(BodyPart part, StringBuilder htmlContent, Map<String, String> imageMap) throws Exception {
        if (part.getContent() instanceof Multipart) {
            Multipart alternative = (Multipart) part.getContent();

            for (int i = 0; i < alternative.getCount(); i++) {
                BodyPart subPart = alternative.getBodyPart(i);

                if (subPart.isMimeType("text/html")) {
                    htmlContent.append(subPart.getContent()).append("<br/>");
                } else if (subPart.isMimeType("image/*")) {
                    processImg(subPart, imageMap);
                } else if (subPart.isMimeType("multipart/alternative") || subPart.getContentType().contains("multipart/ALTERNATIVE")) {
                    processMultipartAlternative(subPart, htmlContent, imageMap);
                } else if (subPart.getContentType().contains("multipart/related") || subPart.getContentType().contains("multipart/RELATED")) {
                    processMultipartAlternative(subPart, htmlContent, imageMap);
                } else {
                    LOGGER.warn("[processMultipartAlternative]其他部分 ContentType: " + subPart.getContentType());
                }
            }
        } else {
            LOGGER.warn("multipart/ALTERNATIVE 的内容不是 Multipart 类型");
        }
    }

    private static String getContent(StringBuilder htmlContent, Map<String, String> imageMap) {
        String content = htmlContent.toString();

        // 如果找到HTML内容和图片，进行替换
        if (!imageMap.isEmpty()) {
            for (Map.Entry<String, String> entry : imageMap.entrySet()) {
                String contentId = entry.getKey();
                String base64Image = entry.getValue();

                // 替换HTML中的图片引用
                String imgPattern = "cid:" + contentId;
                String base64Pattern = "data:image/jpeg;base64," + base64Image;
                content = content.replace(imgPattern, base64Pattern);
            }
        }
        return content;
    }

    private void processImg(BodyPart part, Map<String, String> imageMap) throws MessagingException, IOException {
        // 获取Content-ID
        String[] contentIds = part.getHeader("Content-ID");
        if (contentIds != null && contentIds.length > 0) {
            String contentId = contentIds[0].replaceAll("[<>]", "");
            // 转换图片为base64
            String base64Image = convertImageToBase64(part);
            imageMap.put(contentId, base64Image);
        }
    }

    private String convertImageToBase64(BodyPart part) throws MessagingException, IOException {
        try (InputStream is = part.getInputStream()) {
            ByteArrayOutputStream baos = new ByteArrayOutputStream();
            byte[] buffer = new byte[4096];
            int bytesRead;
            while ((bytesRead = is.read(buffer)) != -1) {
                baos.write(buffer, 0, bytesRead);
            }
            return Base64.getEncoder().encodeToString(baos.toByteArray());
        }
    }

    @Override
    public ExecutionMode getExecutionMode() {
        return ExecutionMode.ASYNCHRONOUS;
    }

    @Override
    public boolean isHandling(Event event) {
        return event instanceof MailboxEvents.Added;
    }

    @Override
    public Group getDefaultGroup() {
        return new UIDEventListenerGroup();
    }

    public static class UIDEventListenerGroup extends Group {
    }
}