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
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87

package org.obm.delivery.cassandra;

import java.util.List;

import org.apache.james.mime4j.dom.Message;
import org.apache.james.mime4j.dom.MessageBuilder;
import org.obm.delivery.cassandra.filters.DeliveryOrder;
import org.obm.delivery.cassandra.filters.IFilterFactory;
import org.obm.delivery.cassandra.filters.IFilterRule;
import org.obm.delivery.cassandra.filters.IRuleEngineImplementation;
import org.obm.delivery.cassandra.filters.RuleEngine;
import org.obm.delivery.cassandra.impl.BlownUpMessage;
import org.obm.delivery.cassandra.impl.CassWriter;
import org.obm.delivery.cassandra.impl.MessageDenormalizer;
import org.obm.lmtp.backend.IDeliveryAgent;
import org.obm.lmtp.backend.LmtpAddress;
import org.obm.lmtp.backend.LmtpEnvelope;
import org.obm.mina.Blob;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * Parses emails using mime4j and stores them in cassandra using bigg model,
 * which is supposed to be optimized for MiniG v2.
 * 
 * @author tom
 * 
 */
public class CassandraAgent implements IDeliveryAgent {

	private static final Logger logger = LoggerFactory
			.getLogger(CassandraAgent.class);

	private IFilterFactory ff;

	private MessageBuilder builder;

	public CassandraAgent(IFilterFactory ff, MessageBuilder builder) {
		this.builder = builder;
		this.ff = ff;
	}

	@Override
	public void deliver(LmtpEnvelope env, Blob blob) {
		long time = System.currentTimeMillis();
		try {
			Message msg = builder.parse(blob.getInputStream());

			List<LmtpAddress> addresses = env.getRecipients();
			IRuleEngineImplementation rei = ff.getEngine();
			for (LmtpAddress addr : addresses) {
				List<IFilterRule> rules = ff.loadRules(addr);
				RuleEngine re = new RuleEngine(addr, rules, rei);

				re.apply(msg);

				DeliveryOrder order = re.getOrder();
				switch (order.getKind()) {
				case FOLDER:
					deliverInto(msg, order.getFolder());
					break;
				case DISCARD:
					logger.info("[" + addr.getEmailAddress()
							+ "] discarded by filter");
					break;
				case REJECT:
					logger.info("[" + addr.getEmailAddress()
							+ "] rejected by filter");
					break;
				}
			}
		} catch (Exception e) {
			logger.error("error delivering msg", e);
		}
		time = System.currentTimeMillis() - time;
		logger.info("delivery time: " + time + "ms.");
	}

	private void deliverInto(Message msg, String folder) {
		logger.info("Deliver " + msg.getMessageId() + " into " + folder);
		MessageDenormalizer denorm = new MessageDenormalizer(msg);
		BlownUpMessage bum = denorm.blowUpInto(folder);
		CassWriter writer = new CassWriter();
		writer.write(bum);
	}

}