/
LightEventSpout.java
174 lines (147 loc) · 4.53 KB
/
LightEventSpout.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
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
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
package movingAverage;
import java.util.Map;
import backtype.storm.spout.SpoutOutputCollector;
import backtype.storm.task.TopologyContext;
import backtype.storm.topology.IRichSpout;
import backtype.storm.topology.OutputFieldsDeclarer;
import backtype.storm.tuple.Fields;
import backtype.storm.tuple.Values;
import java.io.InputStream;
import java.io.OutputStream;
import gnu.io.CommPortIdentifier;
import gnu.io.SerialPort;
import gnu.io.SerialPortEvent;
import gnu.io.SerialPortEventListener;
import java.util.Enumeration;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
public class LightEventSpout implements IRichSpout, SerialPortEventListener {
private static final long serialVersionUID = 1L;
SerialPort serialPort;
/** The port we're normally going to use. */
private static final String PORT_NAMES[] = {
"/dev/tty.usbmodemfa131", // Mac OS X
};
/** Buffered input stream from the port */
private InputStream input;
/** The output stream to the port */
private OutputStream output;
/** Milliseconds to block while waiting for port open */
private static final int TIME_OUT = 2000;
/** Default bits per second for COM port. */
private static final int DATA_RATE = 9600;
private SpoutOutputCollector collector;
private String deviceID = "Arduino";
private BlockingQueue<Integer> blockingQueue = new LinkedBlockingQueue<Integer>();
@Override
public boolean isDistributed() {
return true;
}
@Override
public void open(@SuppressWarnings("rawtypes") final Map conf, final TopologyContext context,
final SpoutOutputCollector collector) {
this.collector = collector;
CommPortIdentifier portId = null;
Enumeration portEnum = CommPortIdentifier.getPortIdentifiers();
// iterate through, looking for the port
while (portEnum.hasMoreElements()) {
CommPortIdentifier currPortId = (CommPortIdentifier) portEnum.nextElement();
for (String portName : PORT_NAMES) {
if (currPortId.getName().equals(portName)) {
portId = currPortId;
System.out.println(portName);
System.out.println(portId);
break;
}
}
}
if (portId == null) {
System.out.println("Could not find COM port.");
return;
}
try {
// open serial port, and use class name for the appName.
System.out.println("serial port : " + serialPort);
serialPort = (SerialPort) portId.open(this.getClass().getName(),
TIME_OUT);
System.out.println("serial port : " + serialPort);
// set port parameters
serialPort.setSerialPortParams(DATA_RATE,
SerialPort.DATABITS_8,
SerialPort.STOPBITS_1,
SerialPort.PARITY_NONE);
// open the streams
input = serialPort.getInputStream();
output = serialPort.getOutputStream();
// add event listeners
serialPort.addEventListener(this);
serialPort.notifyOnDataAvailable(true);
} catch (Exception e) {
System.err.println(e.toString());
}
}
/**
* This should be called when you stop using the port.
* This will prevent port locking on platforms like Linux.
*/
public synchronized void closeSerial() {
if (serialPort != null) {
serialPort.removeEventListener();
serialPort.close();
}
}
@Override
public void nextTuple() {
try {
collector.emit(new Values(deviceID, blockingQueue.take()));
} catch (InterruptedException e) {
// TODO Auto-generated catch block
e.printStackTrace();
}
}
@Override
public void close() {
}
@Override
public void ack(final Object id) {
}
@Override
public void fail(final Object id) {
}
@Override
public void declareOutputFields(final OutputFieldsDeclarer declarer) {
declarer.declare(new Fields("string","double"));
}
@Override
public void serialEvent(SerialPortEvent oEvent) {
if (oEvent.getEventType() == SerialPortEvent.DATA_AVAILABLE) {
try {
int available = input.available();
byte chunk[] = new byte[available];
input.read(chunk, 0, available);
int j = 0;
// System.out.println("chunk length " + chunk.length);
while(j < chunk.length && chunk[j] != '\n') {
j++;
}
j++;
byte number[] = new byte[4];
int count = 0;
while(j < chunk.length-1) {
while (chunk[j] != 10 && chunk[j] != 13 && j < chunk.length-1) {
number[count++] = chunk[j];
j++;
}
j++;
System.out.println(new String(number));
count = 0;
blockingQueue.add(Integer.parseInt(new String(number)));
}
} catch (Exception e) {
System.out.println("Error");
System.err.println(e.toString());
}
}
// Ignore all the other eventTypes, but you should consider the other ones.
}
}