forked from nathanmarz/storm
/
ThriftServer.java
60 lines (51 loc) · 2.07 KB
/
ThriftServer.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
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
package backtype.storm.security.auth;
import java.util.Map;
import javax.security.auth.login.Configuration;
import org.apache.thrift7.TProcessor;
import org.apache.thrift7.server.TServer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import backtype.storm.utils.Utils;
public class ThriftServer {
private Map _storm_conf; //storm configuration
private TProcessor _processor = null;
private int _port = 0;
private TServer _server;
private static final Logger LOG = LoggerFactory.getLogger(ThriftServer.class);
private String _loginConfigurationFile;
public ThriftServer(Map storm_conf, TProcessor processor, int port) {
try {
_storm_conf = storm_conf;
_processor = processor;
_port = port;
_loginConfigurationFile = System.getProperty("java.security.auth.login.config");
if ((_loginConfigurationFile==null) || (_loginConfigurationFile.length()==0)) {
//apply Storm configuration for JAAS login
Map conf = Utils.readStormConfig();
_loginConfigurationFile = (String)conf.get("java.security.auth.login.config");
}
} catch (Exception x) {
x.printStackTrace();
}
}
public void stop() {
if (_server != null)
_server.stop();
}
public void serve() {
try {
//retrieve authentication configuration
Configuration login_conf = AuthUtils.GetConfiguration(_storm_conf);
//locate our thrift transport plugin
ITransportPlugin transportPlugin = AuthUtils.GetTransportPlugin(_storm_conf, login_conf);
//server
_server = transportPlugin.getServer(_port, _processor);
//start accepting requests
_server.serve();
} catch (Exception ex) {
LOG.error("ThriftServer is being stopped due to: " + ex, ex);
if (_server != null) _server.stop();
System.exit(1); //shutdown server process since we could not handle Thrift requests any more
}
}
}