Top
Interstage Big Data Complex Event Processing Server V1.0.0 Developer's Reference
Interstage
Chapter 3 Input Adapter Reference > 3.5 Sample Programs > 3.5.2 HTTP Adapter
PreviousNext

3.5.2 HTTP Adapter

Sample program source code:

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
import java.io.BufferedOutputStream;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStreamReader;
import java.net.HttpURLConnection;
import java.net.MalformedURLException;
import java.net.URL;
 
public class HttpClient {
    URL httpAdapterUrl = null;
    String url = "http://%HOSTNAME%/%ENGINE%FrontServerService/HttpReceiver";
    HttpURLConnection con = null;
    
    public static void main(String[] args)  {
        String hostName = "";
        String engineName = "";
        String dataType = "";
        String charSet = "";
        String eventTypeId = "";
        String data = "";
        long   lWait = 10;
        
        try {
            int loop = 0;
            if (args.length != 8) {
                System.out.println("param is Abnormal");
                return;
            }
            
            hostName    = String.valueOf(args[0]);
            engineName  = String.valueOf(args[1]);
            dataType    = String.valueOf(args[2]);
            charSet     = String.valueOf(args[3]);
            eventTypeId = String.valueOf(args[4]);
            data        = String.valueOf(args[5]);
            lWait       = Long.valueOf(args[6]);
            loop        = Integer.valueOf(args[7]);
            
            HttpClient hc = new HttpClient(hostName, engineName);
            hc.sendMessage(data, charSet, dataType, eventTypeId, loop, lWait);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    
    public HttpClient(String hostName, String engineName) throws IOException {
        
        try {
            String wkUrl = url.replaceAll("%ENGINE%", engineName);
            wkUrl = wkUrl.replaceAll("%HOSTNAME%", hostName);
            
            httpAdapterUrl = new URL(wkUrl);
        } catch (MalformedURLException e) {
            e.printStackTrace();
        }
    }
    
    private HttpURLConnection open(String charSet, String dataType, String eventTypeId)
            throws Exception {
        
        HttpURLConnection con = null;
        con = (HttpURLConnection) httpAdapterUrl.openConnection();
        
        con.setRequestMethod("POST");
        con.setRequestProperty("TYPE", dataType);
        con.setRequestProperty("EVENT-TYPE-ID", eventTypeId);
        
        if ("csv".equalsIgnoreCase(dataType)) {
            con.setRequestProperty("content-type", "text/plain; charset=" + charSet);
        } else if ("xml".equalsIgnoreCase(dataType)) {
            con.setRequestProperty("content-type", "text/xml; charset=" + charSet);
        } else {
            System.out.println("datatype is Abnormal");
            throw new Exception();
        }
        
        con.setDoOutput(true);
        con.connect();
        
        return con;
    }
    
    private void sendMessage(String data, String charSet, String dataType, 
        String eventTypeId,int loop, long lWait) throws Exception {
        
        BufferedOutputStream bos = null;
        InputStreamReader ir1 = null;
        BufferedReader br1 = null;
        
        try {
            for (int i = 0; i < loop; i++) {
                con = this.open(charSet, dataType, eventTypeId);
                bos = new BufferedOutputStream(con.getOutputStream());
                
                String MSG = data.replaceAll("%COUNTER%", String.valueOf(i));
                bos.write(MSG.getBytes(charSet));
                
                bos.flush();
                bos.close();
                
                ir1 = new InputStreamReader(con.getInputStream());
                br1 = new BufferedReader(ir1);
                
                String line;
                while ((line = br1.readLine()) != null) {
                    System.out.println(line);
                }
                
                br1.close();
                ir1.close();
                
                con.disconnect();
                Thread.sleep(lWait);
            }
            
        } catch (MalformedURLException e) {
            e.printStackTrace();
            throw e;
        } catch (IOException e) {
            e.printStackTrace();
            throw e;
        } catch (InterruptedException e) {
            e.printStackTrace();
            throw e;
        } finally {
            try {
                if (br1 != null) {
                    br1.close();
                }
                if (ir1 != null) {
                    ir1.close();
                }
            } catch (IOException e) {
                e.printStackTrace();
                return;
            }
        }
        
        return;
    }
}

The following data are obtained from the arguments at runtime:

Argument

Variable

Use

1

hostName

CEP Server host name

2

engineName

CEP engine name

3

dataType

Event data format

4

charSet

Character set

5

eventTypeId

Event type ID

6

data

Event data
%COUNTER% included in event data needs to be substituted with loop count so that different event data shall be sent at each loop.

7

lWait

Data transmission wait time

8

loop

Data transmission count

3.5.2.1 Example of Sample Execution(Sends a CSV data)

An example of sample execution is shown below.

In this example, debug information is output to the engine log by using DebugLogListener.

Command execution result

# java -cp ./ HttpClient localhost CepEngine CSV UTF-8 CSVEvent HTTP,CSV,%COUNTER% 1 3 <ENTER>
Code=0 Message=Sending message completed normally.
Code=0 Message=Sending message completed normally.
Code=0 Message=Sending message completed normally.

Engine log output result

2012-07-29 13:05:02,954 [DEBUG] abc:length=1
abc[0]
        operation       :CSV: String
        count   :0: String
        ID      :HTTP: String
 
2012-07-29 13:05:03,027 [DEBUG] abc:length=1
abc[0]
        operation       :CSV: String
        count   :1: String
        ID      :HTTP: String
 
2012-07-29 13:05:03,108 [DEBUG] abc:length=1
abc[0]
        operation       :CSV: String
        count   :2: String
        ID      :HTTP: String

3.5.2.2 Example of Sample Execution(Sends an XML data)

An example of sample execution is shown below.

In this example, debug information is output to the engine log by using DebugLogListener.

Command execution result

# java -cp ./ HttpClient localhost CepEngine XML UTF-8 XMLEvent '<?xml version="1.0"\
 encoding="UTF-8"?><XMLEvent><ID>HTTP</ID> <operation>XML</operation>\
 <count>%COUNTER%</count></XMLEvent>' 1 3 <ENTER>
Code=0 Message=Sending message completed normally.
Code=0 Message=Sending message completed normally.
Code=0 Message=Sending message completed normally.

Please note that we added backslash ("\") and newline characters in the command line example above for readability only - the actual command line does not have a backslash and newline.

Engine log output result

2012-07-29 13:07:32,670 [DEBUG] abc--0:length=1
abc--0[0]
        operation       :XML: String
        count   :0: String
        ID      :HTTP: String
 
2012-07-29 13:07:32,685 [DEBUG] abc--0:length=1
abc--0[0]
        operation       :XML: String
        count   :1: String
        ID      :HTTP: String
 
2012-07-29 13:07:32,698 [DEBUG] abc--0:length=1
abc--0[0]
        operation       :XML: String
        count   :2: String
        ID      :HTTP: String
PreviousNext
Copyright 2012 FUJITSU LIMITED