Skip to content

Commit 5615670

Browse files
author
chengyitian
committed
Merge remote-tracking branch 'origin/dev' into dev
2 parents 515db1a + 8bd4810 commit 5615670

File tree

2 files changed

+215
-159
lines changed

2 files changed

+215
-159
lines changed

‎test/com/xxdb/streaming/client/cep/EventClientTest.java‎

Lines changed: 94 additions & 38 deletions
Original file line numberDiff line numberDiff line change
@@ -8,7 +8,6 @@
88
import org.junit.Test;
99

1010
import java.io.IOException;
11-
import java.math.BigInteger;
1211
import java.time.LocalDateTime;
1312
import java.time.LocalTime;
1413
import java.util.*;
@@ -129,7 +128,7 @@ public static void Preparedata_array(long count1,long count2) throws IOException
129128
"cdecimal32 = array(DECIMAL32(2)[]).append!(cut(decimal32(take(-100..100 join NULL, n) + 0.254, 3), m))\n" +
130129
"cdecimal64 = array(DECIMAL64(7)[]).append!(cut(decimal64(take(-100..100 join NULL, n) + 0.25467, 4), m))\n" +
131130
"cdecimal128 = array(DECIMAL128(19)[]).append!(cut(decimal128(take(-100..100 join NULL, n) + 0.25467, 5), m))\n" +
132-
"data = table(cbool, cchar, cshort, cint, clong, cdouble, cfloat, cdate, cmonth, ctime, cminute, csecond, cdatetime, ctimestamp, cnanotime, cnanotimestamp, cdatehour, cuuid, cipaddr, cint128, cpoint, ccomplex, cdecimal32, cdecimal64)\n" ;
131+
"share table(cbool, cchar, cshort, cint, clong, cdouble, cfloat, cdate, cmonth, ctime, cminute, csecond, cdatetime, ctimestamp, cnanotime, cnanotimestamp, cdatehour, cuuid, cipaddr, cint128, cpoint, ccomplex, cdecimal32, cdecimal64,cdecimal128) as data;" ;
133132
conn.run(script1);
134133
}
135134
public static void PrepareUser(String userName,String password) throws IOException {
@@ -225,27 +224,6 @@ public void test_EventClient_EventType_null_1() throws IOException, Interrupted
225224
}
226225
Assert.assertEquals("eventType must be non-empty.",re);
227226
}
228-
229-
@Test//到时候修改成覆盖所有数据类型的
230-
public void test_EventClient_EventType_special_character() throws IOException, InterruptedException {
231-
String script = "share streamTable(1000000:0, `time`eventType`event, [TIMESTAMP,STRING,BLOB]) as inputTable;\n"+
232-
"share table(100:0, `market`code`price`qty`eventTime, [STRING, STRING, DOUBLE, INT, TIMESTAMP]) as outputTable;\n";
233-
//conn.run(script);
234-
EventScheme scheme = new EventScheme();
235-
scheme.setEventType("MarketData");
236-
scheme.setAttrKeys(Arrays.asList("market", "code", "price", "qty", "eventTime"));
237-
scheme.setAttrTypes(Arrays.asList(DT_STRING, DT_STRING, DT_DOUBLE, DT_INT, DT_TIMESTAMP));
238-
scheme.setAttrForms(Arrays.asList(DF_SCALAR, DF_SCALAR, DF_SCALAR, DF_SCALAR, DF_SCALAR));
239-
List<EventScheme> eventSchemes = Collections.singletonList(scheme);
240-
List<String> eventTimeKeys = Collections.singletonList("eventTime");
241-
List<String> commonKeys = new ArrayList<>();
242-
EventClient client = new EventClient(eventSchemes, eventTimeKeys, commonKeys);
243-
client.subscribe(HOST, PORT, "outputTable", "test1", handler, -1, true, "admin", "123456");
244-
Thread.sleep(200000);
245-
BasicTable bt1 = (BasicTable)conn.run("select * from outputTable;");
246-
Assert.assertEquals(1,bt1.rows());
247-
}
248-
249227
@Test
250228
public void test_EventClient_EventType_repetition() throws IOException, InterruptedException {
251229
EventScheme scheme = new EventScheme();
@@ -1304,47 +1282,125 @@ public void test_EventClient_subscribe_all_dateType_100() throws IOException, I
13041282

13051283
@Test
13061284
public void test_EventClient_all_dateType_vector() throws IOException, InterruptedException {
1307-
//DBConnection conn = new DBConnection(true);
1308-
//conn.connect(HOST, PORT, "admin", "123456");
1285+
Preparedata_array(100,10);
13091286
String script = "share streamTable(1000000:0, `eventType`event, [STRING,BLOB]) as inputTable;\n"+
1310-
"colNames=\"col\"+string(1..24);\n" +
1311-
"colTypes=[BOOL[],CHAR[],SHORT[],INT[],LONG[],DOUBLE[],FLOAT[],DATE[],MONTH[],TIME[],MINUTE[],SECOND[],DATETIME[],TIMESTAMP[],NANOTIME[],NANOTIMESTAMP[],DATEHOUR[],UUID[],IPADDR[],INT128[],POINT[],COMPLEX[],DECIMAL32(2)[],DECIMAL64(7)[]];\n" +
1287+
"colNames=\"col\"+string(1..25);\n" +
1288+
"colTypes=[BOOL[],CHAR[],SHORT[],INT[],LONG[],DOUBLE[],FLOAT[],DATE[],MONTH[],TIME[],MINUTE[],SECOND[],DATETIME[],TIMESTAMP[],NANOTIME[],NANOTIMESTAMP[], DATEHOUR[],UUID[],IPADDR[],INT128[],POINT[],COMPLEX[],DECIMAL32(2)[],DECIMAL64(7)[],DECIMAL128(10)[]];\n" +
13121289
"share table(1:0,colNames,colTypes) as outputTable;\n" ;
13131290
conn.run(script);
1314-
1291+
String script1 ="class event_all_array_dateType{\n" +
1292+
"\tboolv :: BOOL VECTOR\n" +
1293+
"\tcharv :: CHAR VECTOR\n" +
1294+
"\tshortv :: SHORT VECTOR\n" +
1295+
"\tintv :: INT VECTOR\n" +
1296+
"\tlongv :: LONG VECTOR\n" +
1297+
"\tdoublev :: DOUBLE VECTOR\n" +
1298+
"\tfloatv :: FLOAT VECTOR\n" +
1299+
"\tdatev :: DATE VECTOR\n" +
1300+
"\tmonthv :: MONTH VECTOR\n" +
1301+
"\ttimev :: TIME VECTOR\n" +
1302+
"\tminutev :: MINUTE VECTOR\n" +
1303+
"\tsecondv :: SECOND VECTOR\n" +
1304+
"\tdatetimev :: DATETIME VECTOR \n" +
1305+
"\ttimestampv :: TIMESTAMP VECTOR\n" +
1306+
"\tnanotimev :: NANOTIME VECTOR\n" +
1307+
"\tnanotimestampv :: NANOTIMESTAMP VECTOR\n" +
1308+
"\t//stringv :: STRING VECTOR\n" +
1309+
"\tdatehourv :: DATEHOUR VECTOR\n" +
1310+
"\tuuidv :: UUID VECTOR\n" +
1311+
"\tippaddrv :: IPADDR VECTOR\n" +
1312+
"\tint128v :: INT128 VECTOR\n" +
1313+
"\t//blobv :: BLOB VECTOR\n" +
1314+
"\tpointv :: POINT VECTOR\n" +
1315+
"\tcomplexv :: COMPLEX VECTOR\n" +
1316+
"\tdecimal32v :: DECIMAL32(3) VECTOR\n" +
1317+
"\tdecimal64v :: DECIMAL64(8) VECTOR\n" +
1318+
"\tdecimal128v :: DECIMAL128(10) VECTOR \n" +
1319+
" def event_all_array_dateType(bool, char, short, int, long, double, float, date, month, time, minute, second, datetime, timestamp, nanotime, nanotimestamp, datehour, uuid, ippaddr, int128,point, complex, decimal32, decimal64, decimal128){\n" +
1320+
"\tboolv = bool\n" +
1321+
"\tcharv = char\n" +
1322+
"\tshortv = short\n" +
1323+
"\tintv = int\n" +
1324+
"\tlongv = long\n" +
1325+
"\tdoublev = double\n" +
1326+
"\tfloatv = float\n" +
1327+
"\tdatev = date\n" +
1328+
"\tmonthv = month\n" +
1329+
"\ttimev = time\n" +
1330+
"\tminutev = minute\n" +
1331+
"\tsecondv = second\n" +
1332+
"\tdatetimev = datetime\n" +
1333+
"\ttimestampv = timestamp\n" +
1334+
"\tnanotimev = nanotime\n" +
1335+
"\tnanotimestampv = nanotimestamp\n" +
1336+
"\t//stringv = string\n" +
1337+
"\tdatehourv = datehour\n" +
1338+
"\tuuidv = uuid\n" +
1339+
"\tippaddrv = ippaddr\n" +
1340+
"\tint128v = int128\n" +
1341+
"\t//blobv = blob\n" +
1342+
"\tpointv = point\n" +
1343+
"\tcomplexv = complex\n" +
1344+
"\tdecimal32v = decimal32\n" +
1345+
"\tdecimal64v = decimal64\n" +
1346+
"\tdecimal128v = decimal128\n" +
1347+
" \t}\n" +
1348+
"} \n" +
1349+
"schemaTable = table(array(STRING, 0) as eventType, array(STRING, 0) as eventKeys, array(INT[], ) as type, array(INT[], 0) as form)\n" +
1350+
"eventType = 'event_all_dateType'\n" +
1351+
"eventKeys = 'boolv,charv,shortv,intv,longv,doublev,floatv,datev,monthv,timev,minutev,secondv,datetimev,timestampv,nanotimev,nanotimestampv,datehourv,uuidv,ippaddrv,int128v,pointv,complexv,decimal32v,decimal64v,decimal128v';\n" +
1352+
"typeV = [BOOL[], CHAR[], SHORT[], INT[], LONG[], DOUBLE[], FLOAT[], DATE[],MONTH[], TIME[], MINUTE[], SECOND[], DATETIME[], TIMESTAMP[], NANOTIME[], NANOTIMESTAMP[], DATEHOUR[], UUID[], IPADDR[], INT128[], POINT[], COMPLEX[], DECIMAL32(3)[], DECIMAL64(8)[], DECIMAL128(10)[]];\n" +
1353+
"formV = [VECTOR, VECTOR, VECTOR, VECTOR, VECTOR, VECTOR, VECTOR, VECTOR, VECTOR, VECTOR, VECTOR, VECTOR, VECTOR, VECTOR, VECTOR, VECTOR, VECTOR, VECTOR, VECTOR, VECTOR, VECTOR, VECTOR, VECTOR, VECTOR, VECTOR];\n" +
1354+
"insert into schemaTable values([eventType], [eventKeys], [typeV],[formV]);\n" +
1355+
"share streamTable( array(STRING, 0) as eventType, array(BLOB, 0) as blobs) as intput1;\n" +
1356+
"try{\ndropStreamEngine(`serInput)\n}catch(ex){\n}\n" +
1357+
"inputSerializer = streamEventSerializer(name=`serInput, eventSchema=schemaTable, outputTable=intput1);";
1358+
conn.run(script1);
13151359
EventScheme scheme = new EventScheme();
13161360
scheme.setEventType("event_all_array_dateType");
1317-
scheme.setAttrKeys(Arrays.asList("boolv", "charv", "shortv", "intv", "longv", "doublev", "floatv", "datev", "monthv", "timev", "minutev", "secondv", "datetimev", "timestampv", "nanotimev", "nanotimestampv", "datehourv", "uuidv", "ippaddrv", "int128v", "pointv", "complexv", "decimal32v", "decimal64v"));
1318-
scheme.setAttrTypes(Arrays.asList(DT_BOOL, DT_BYTE, DT_SHORT, DT_INT, DT_LONG, DT_DOUBLE, DT_FLOAT, DT_DATE,DT_MONTH, DT_TIME, DT_MINUTE, DT_SECOND, DT_DATETIME, DT_TIMESTAMP, DT_NANOTIME, DT_NANOTIMESTAMP, DT_DATEHOUR, DT_UUID, DT_IPADDR, DT_INT128, DT_POINT, DT_COMPLEX, DT_DECIMAL32, DT_DECIMAL64));
1319-
scheme.setAttrForms(Arrays.asList( DF_VECTOR, DF_VECTOR, DF_VECTOR, DF_VECTOR, DF_VECTOR, DF_VECTOR, DF_VECTOR, DF_VECTOR, DF_VECTOR, DF_VECTOR, DF_VECTOR, DF_VECTOR, DF_VECTOR, DF_VECTOR, DF_VECTOR, DF_VECTOR, DF_VECTOR, DF_VECTOR, DF_VECTOR, DF_VECTOR, DF_VECTOR, DF_VECTOR, DF_VECTOR, DF_VECTOR));
1361+
scheme.setAttrKeys(Arrays.asList("boolv", "charv", "shortv", "intv", "longv", "doublev", "floatv", "datev", "monthv", "timev", "minutev", "secondv", "datetimev", "timestampv", "nanotimev", "nanotimestampv", "datehourv", "uuidv", "ippaddrv", "int128v", "pointv", "complexv", "decimal32v", "decimal64v", "decimal128v"));
1362+
scheme.setAttrTypes(Arrays.asList(DT_BOOL, DT_BYTE, DT_SHORT, DT_INT, DT_LONG, DT_DOUBLE, DT_FLOAT, DT_DATE,DT_MONTH, DT_TIME, DT_MINUTE, DT_SECOND, DT_DATETIME, DT_TIMESTAMP, DT_NANOTIME, DT_NANOTIMESTAMP, DT_DATEHOUR, DT_UUID, DT_IPADDR, DT_INT128, DT_POINT, DT_COMPLEX, DT_DECIMAL32, DT_DECIMAL64, DT_DECIMAL128));
1363+
scheme.setAttrForms(Arrays.asList( DF_VECTOR, DF_VECTOR, DF_VECTOR, DF_VECTOR, DF_VECTOR, DF_VECTOR, DF_VECTOR, DF_VECTOR, DF_VECTOR, DF_VECTOR, DF_VECTOR, DF_VECTOR, DF_VECTOR, DF_VECTOR, DF_VECTOR, DF_VECTOR, DF_VECTOR, DF_VECTOR, DF_VECTOR, DF_VECTOR, DF_VECTOR, DF_VECTOR, DF_VECTOR, DF_VECTOR, DF_VECTOR));
13201364
List<EventScheme> eventSchemes = Collections.singletonList(scheme);
13211365
List<String> eventTimeKeys = new ArrayList<>();
13221366
List<String> commonKeys = new ArrayList<>();
13231367
EventSender sender = EventSender.createEventSender(eventSchemes, eventTimeKeys, commonKeys);
13241368
sender.connect(conn,"inputTable");
1325-
Preparedata_array(100,10);
1326-
BasicTable bt = (BasicTable)conn.run("select * from data");
13271369

13281370
EventClient client = new EventClient(eventSchemes, eventTimeKeys, commonKeys);
1329-
client.subscribe(HOST, PORT, "inputTable", "test1", handler, -1, true, "admin", "123456");
1371+
client.subscribe(HOST, PORT, "intput1", "test1", handler, -1, true, "admin", "123456");
13301372

1373+
Preparedata_array(100,10);
1374+
BasicTable bt = (BasicTable)conn.run("select * from data");
1375+
String script2 = "data1=select * from data;\n" +
1376+
"for(i in 0..9){\n" +
1377+
"\tevent_all_array_dateType1=event_all_array_dateType(data1.row(i)[`cbool], data1.row(i)[`cchar], data1.row(i)[`cshort], data1.row(i)[`cint],data1.row(i)[`clong], data1.row(i)[`cdouble], data1.row(i)[`cfloat], data1.row(i)[`cdate], data1.row(i)[`cmonth], data1.row(i)[`ctime], data1.row(i)[`cminute], data1.row(i)[`csecond], data1.row(i)[`cdatetime], data1.row(i)[`ctimestamp], data1.row(i)[`cnanotime], data1.row(i)[`cnanotimestamp], data1.row(i)[`cdatehour], data1.row(i)[`cuuid], data1.row(i)[`cipaddr], data1.row(i)[`cint128],data1.row(i)[`cpoint], data1.row(i)[`ccomplex], data1.row(i)[`cdecimal32], data1.row(i)[`cdecimal64], data1.row(i)[`cdecimal128])\n" +
1378+
"\tappendEvent(inputSerializer, event_all_array_dateType1)\n" +
1379+
"\t}" ;
1380+
conn.run(script2);
1381+
Thread.sleep(10000);
13311382
for(int i=0;i<bt.rows();i++){
13321383
List<Entity> attributes = new ArrayList<>();
13331384
for(int j=0;j<bt.columns();j++){
13341385
Entity pt = bt.getColumn(j).get(i);
1335-
//System.out.println(pt.getDataType());
1336-
//System.out.println(i + "行， " + j + "列：" + pt.getString());
1386+
System.out.println(pt.getDataType());
1387+
System.out.println(i + "行， " + j + "列：" + pt.getString());
13371388
attributes.add(pt);
13381389
}
13391390
sender.sendEvent("event_all_array_dateType",attributes);
13401391
}
1392+
Thread.sleep(1000);
13411393
BasicTable bt1 = (BasicTable)conn.run("select * from inputTable;");
13421394
Assert.assertEquals(10,bt1.rows());
1343-
Thread.sleep(20000);
1344-
BasicTable bt2 = (BasicTable)conn.run("select * from outputTable;");
1395+
BasicTable bt2 = (BasicTable)conn.run("select * from intput;");
13451396
Assert.assertEquals(10,bt2.rows());
13461397
checkData(bt,bt2);
1398+
Thread.sleep(10000);
1399+
BasicTable bt3 = (BasicTable)conn.run("select * from outputTable;");
1400+
Assert.assertEquals(10,bt3.rows());
1401+
checkData(bt,bt3);
13471402
}
1403+
13481404
public static EventMessageHandler handler_array = new EventMessageHandler() {
13491405
@Override
13501406
public void doEvent(String eventType, List<Entity> attribute) {

0 commit comments

Comments
 (0)