james源码解析（二）

因个人需要，在个人服务器上搭建了邮箱服务器，使用的是mailu，整体基于docker多容器搭建。

本来用起来没什么大问题，但是因为其自带nginx容器，和我原本部署的nginx容器会有一定的冲突，不是很满意

一番查找后，让我找到了james——apache开发的基于java的邮箱服务器。

上一篇章中实现了james的运行和测试，这一篇开始源码入门

源码入门

在 server/Overview.md 中有对项目一些结构性描述，奈何文档只写了个开头，总共不过一百行文字，而且和现有项目都有些对不上了，希望apache社区能早日补充文档

依照我目前的理解，james总体可以分成以下几个部分

  • protocols 协议实现和通信，基于netty的网络通信，实现了协议解析
  • store 数据存储
  • mailet james自已定义的组件名称，我的理解是邮件的处理器，james的各种功能也是通过mailet实现的
  • event 事件机制

当然james不止这点东西，其他的要么我还没了解到，要么没必要深究

想要了解一个项目，首先需要从程序入口开始。

以server/apps/jpa-app/src/main/java/org/apache/james/JPAJamesServerMain.java为例

查看他的main方法

1
2
3
4
5
6
7
8
9
10
11
12
13
public static void main(String[] args) throws Exception {
    ExtraProperties.initialize();

    JPAJamesConfiguration configuration = JPAJamesConfiguration.builder()
        .useWorkingDirectoryEnvProperty()
        .build();

    LOGGER.info("Loading configuration {}", configuration.toString());
    GuiceJamesServer server = createServer(configuration)
        .combineWith(new JMXServerModule());

    JamesServerMain.main(server);
}

头两行代码看起来是配置文件的读取，可以先跳过。重点是最后两行代码。

JamesServerMain.main(server);是调用服务启动方法和注册关闭hook，不重要。

因此看createServer

1
2
3
4
5
6
static GuiceJamesServer createServer(JPAJamesConfiguration configuration) {
    return GuiceJamesServer.forConfiguration(configuration)
        .combineWith(JPA_MODULE_AGGREGATE)
        .combineWith(new UsersRepositoryModuleChooser(new JPAUsersRepositoryModule())
            .chooseModules(configuration.getUsersRepositoryImplementation()));
}

这个combineWith是干什么的？看一下JPA_MODULE_AGGREGATE参数是什么

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
private static final Module JPA_SERVER_MODULE = Modules.combine(
    new ActiveMQQueueModule(),
    new DefaultProcessorsConfigurationProviderModule(),
    new ElasticSearchMetricReporterModule(),
    new JPADataModule(),
    new JPAMailboxModule(),
    new MailboxModule(),
    new LuceneSearchMailboxModule(),
    new NoJwtModule(),
    new RawPostDequeueDecoratorModule(),
    new SieveJPARepositoryModules(),
    new DefaultEventModule(),
    new TaskManagerModule(),
    new MemoryDeadLetterModule(),
    new SpamAssassinListenerModule());

private static final Module JPA_MODULE_AGGREGATE = Modules.combine(
    new MailetProcessingModule(), JPA_SERVER_MODULE, PROTOCOLS);

从命令看出，这应该是在组装应用模块。james将功能拆分开，最后通过组合不同的模块，实现最终提供不同功能的版本

一路深入下去，发现这一套 module 定义是 Guice 提供的，查阅资料可知，这是一个依赖注入框架，类似于Spring，在学习阶段直接当成Spring看待就行。

以server/container/guice/protocols/smtp/src/main/java/org/apache/james/modules/protocols/SMTPServerModule.java 为例

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
public class SMTPServerModule extends AbstractModule {
    @Override
    protected void configure() {
        install(new JSPFModule());
        bind(SMTPServerFactory.class).in(Scopes.SINGLETON);
        bind(OioSMTPServerFactory.class).in(Scopes.SINGLETON);

        Multibinder.newSetBinder(binder(), GuiceProbe.class).addBinding().to(SmtpGuiceProbe.class);
    }

    @ProvidesIntoSet
    InitializationOperation configureSmtp(ConfigurationProvider configurationProvider,
                                        SMTPServerFactory smtpServerFactory,
                                        SendMailHandler sendMailHandler) {
        return InitilizationOperationBuilder
            .forClass(SMTPServerFactory.class)
            .init(() -> {
                smtpServerFactory.configure(configurationProvider.getConfiguration("smtpserver"));
                smtpServerFactory.init();
                sendMailHandler.init(null);
            });
    }

}

核心方法应该是 smtpServerFactory.init() ，一路追踪下去，最终来到了server/protocols/protocols-library/src/main/java/org/apache/james/protocols/lib/netty/AbstractConfigurableAsyncServer#init

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
@PostConstruct
public final void init() throws Exception {

    if (isEnabled()) {

        buildSSLContext();
        preInit();
        executionHandler = createExecutionHandler();
        frameHandlerFactory = createFrameHandlerFactory();
        bind();
        port = retrieveFirstBindedPort();

        mbeanServer = ManagementFactory.getPlatformMBeanServer();
        registerMBean();
        
        LOGGER.info("Init {} done", getServiceType());

    }

}

这个模块是SMTP协议实现，james的网络通信是使用的netty，netty核心是handler，所以需要关注的就是Handler实现类。

结合之前测试的日志可以发现有个SMTPChannelUpstreamHandler类，排查代码，定位到 SMTPServer的第221行。

这里的实现逻辑是抽象类提供方法负责注册handler，子类重写方法提供不同的handler实现。典型的模板模式

SMTPChannelUpstreamHandler三个核心方法，也就是netty提供的方法，分别对应连接建立，收到消息和连接关闭。

查看代码可知，具体实现在其父类BasicChannelUpstreamHandler

先看连接建立

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
public void channelConnected(ChannelHandlerContext ctx, ChannelStateEvent e) throws Exception {
    try (Closeable closeable = mdc(ctx).build()) {
        List<ConnectHandler> connectHandlers = chain.getHandlers(ConnectHandler.class);
        List<ProtocolHandlerResultHandler> resultHandlers = chain.getHandlers(ProtocolHandlerResultHandler.class);
        ProtocolSession session = (ProtocolSession) ctx.getAttachment();
        LOGGER.info("Connection established from {}", session.getRemoteAddress().getAddress().getHostAddress());
        if (connectHandlers != null) {
            for (ConnectHandler cHandler : connectHandlers) {
                long start = System.currentTimeMillis();
                Response response = cHandler.onConnect(session);
                long executionTime = System.currentTimeMillis() - start;

                for (ProtocolHandlerResultHandler resultHandler : resultHandlers) {
                    resultHandler.onResponse(session, response, executionTime, cHandler);
                }
                if (response != null) {
                    // TODO: This kind of sucks but I was able to come up with something more elegant here
                    ((ProtocolSessionImpl) session).getProtocolTransport().writeResponse(response, session);
                }

            }
        }
        super.channelConnected(ctx, e);
    }
}

其实三个方法逻辑都大差不大，都是通过chain获取handler，依次执行后写入数据到连接。只是不同事件封装了不同的Handler接口

ProtocolHandlerChain chain里维护了一个handler的列表，查看其子类找到SMTPProtocolHandlerChain#initDefaultHandlers，可以看到其维护的handler列表。此时出现问题了，通过ide无法找到该类构造方法调用位置，debug断点也没有执行，反倒找到了另外一个类org.apache.james.smtpserver.CoreCmdHandlerLoader,这个类同样维护了一个handler列表，这就有点迷惑了，这到底是怎么个流程。

实在搞不懂，先暂时跳过，总之知道了有哪些handler。

回到BasicChannelUpstreamHandler，重点看消息接受方法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
public void messageReceived(ChannelHandlerContext ctx, MessageEvent e) throws Exception {
    try (Closeable closeable = mdc(ctx).build()) {
        ProtocolSession pSession = (ProtocolSession) ctx.getAttachment();
        LinkedList<LineHandler> lineHandlers = chain.getHandlers(LineHandler.class);
        LinkedList<ProtocolHandlerResultHandler> resultHandlers = chain.getHandlers(ProtocolHandlerResultHandler.class);


        if (lineHandlers.size() > 0) {

            ChannelBuffer buf = (ChannelBuffer) e.getMessage();
            LineHandler lHandler = (LineHandler) lineHandlers.getLast();
            long start = System.currentTimeMillis();
            Response response = lHandler.onLine(pSession, buf.toByteBuffer());
            long executionTime = System.currentTimeMillis() - start;

            for (ProtocolHandlerResultHandler resultHandler : resultHandlers) {
                response = resultHandler.onResponse(pSession, response, executionTime, lHandler);
            }
            if (response != null) {
                // TODO: This kind of sucks but I was able to come up with something more elegant here
                ((ProtocolSessionImpl) pSession).getProtocolTransport().writeResponse(response, pSession);
            }

        }

        super.messageReceived(ctx, e);
    }
}

这里可以看到有点不太一样了，获取的LineHandler列表只执行了最后一个，这是为何啊。

从之前的handler列表中，可以看到CommandDispatcher，从名字看是服务命令分发的。其他handler也基本是各个命令的实现。

看到这里已经大致明白了网络协议的部分架构，但是还没搞懂如何和核心的邮件处理关联

后端 > java > james
#java #james
james源码解析（二）
http://blog.inkroom.cn/2022/11/13/3X95FN1.html
作者
inkbox
发布于
2022年11月13日
许可协议