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
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
|
/*
* Copyright 2012 - 2015 Anton Tananaev (anton.tananaev@gmail.com)
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
package org.traccar;
import java.net.InetSocketAddress;
import org.jboss.netty.buffer.ChannelBuffer;
import org.jboss.netty.buffer.ChannelBuffers;
import org.jboss.netty.channel.ChannelEvent;
import org.jboss.netty.channel.ChannelHandlerContext;
import org.jboss.netty.channel.ChannelPipeline;
import org.jboss.netty.channel.ChannelPipelineFactory;
import org.jboss.netty.channel.ChannelStateEvent;
import org.jboss.netty.channel.Channels;
import org.jboss.netty.channel.DownstreamMessageEvent;
import org.jboss.netty.channel.ExceptionEvent;
import org.jboss.netty.channel.MessageEvent;
import org.jboss.netty.channel.SimpleChannelHandler;
import org.jboss.netty.handler.logging.LoggingHandler;
import org.jboss.netty.handler.timeout.IdleStateHandler;
import org.traccar.helper.Log;
public abstract class BasePipelineFactory implements ChannelPipelineFactory {
private final TrackerServer server;
private FilterHandler filterHandler;
private Integer resetDelay;
private Boolean processInvalidPositions;
protected class OpenChannelHandler extends SimpleChannelHandler {
private final TrackerServer server;
public OpenChannelHandler(TrackerServer server) {
this.server = server;
}
@Override
public void channelOpen(ChannelHandlerContext ctx, ChannelStateEvent e) {
server.getChannelGroup().add(e.getChannel());
}
}
/**
* Logging using global logger
*/
protected class StandardLoggingHandler extends LoggingHandler {
@Override
public void log(ChannelEvent e) {
if (e instanceof MessageEvent) {
MessageEvent event = (MessageEvent) e;
StringBuilder msg = new StringBuilder();
msg.append("[").append(String.format("%08X", e.getChannel().getId())).append(": ");
msg.append(((InetSocketAddress) e.getChannel().getLocalAddress()).getPort());
msg.append((e instanceof DownstreamMessageEvent) ? " > " : " < ");
msg.append(((InetSocketAddress) event.getRemoteAddress()).getAddress().getHostAddress()).append("]");
// Append hex message
if (event.getMessage() instanceof ChannelBuffer) {
msg.append(" HEX: ");
msg.append(ChannelBuffers.hexDump((ChannelBuffer) event.getMessage()));
}
Log.debug(msg.toString());
} else if (e instanceof ExceptionEvent) {
ExceptionEvent event = (ExceptionEvent) e;
Log.warning(event.getCause());
}
}
}
public BasePipelineFactory(TrackerServer server, String protocol) {
this.server = server;
String resetDelayProperty = Context.getProps().getProperty(protocol + ".resetDelay");
if (resetDelayProperty != null) {
resetDelay = Integer.valueOf(resetDelayProperty);
}
String enableFilter = Context.getProps().getProperty("filter.enable");
if (enableFilter != null && Boolean.valueOf(enableFilter)) {
filterHandler = new FilterHandler();
}
if (Context.getReverseGeocoder() != null) {
// Default behavior is to process invalid positions (i.e., the "null" case)
String invalidPositions = Context.getProps().getProperty("geocode.processInvalidPositions");
processInvalidPositions = (invalidPositions == null || Boolean.valueOf(invalidPositions));
}
}
protected abstract void addSpecificHandlers(ChannelPipeline pipeline);
@Override
public ChannelPipeline getPipeline() {
ChannelPipeline pipeline = Channels.pipeline();
if (resetDelay != null) {
pipeline.addLast("idleHandler", new IdleStateHandler(GlobalTimer.getTimer(), resetDelay, 0, 0));
}
pipeline.addLast("openHandler", new OpenChannelHandler(server));
if (Context.isLoggerEnabled()) {
pipeline.addLast("logger", new StandardLoggingHandler());
}
addSpecificHandlers(pipeline);
if (filterHandler != null) {
pipeline.addLast("filter", filterHandler);
}
if (Context.getReverseGeocoder() != null) {
pipeline.addLast("geocoder", new ReverseGeocoderHandler(Context.getReverseGeocoder(), processInvalidPositions));
}
pipeline.addLast("remoteAddress", new RemoteAddressHandler());
pipeline.addLast("handler", new MainEventHandler());
return pipeline;
}
}
|
