新增异步查询接口，优化数据库存储
This commit is contained in:
ascrutae
parent
e8eef8e0a8
commit
648e2d8b18
|
|
@ -3,20 +3,44 @@ package com.a.eye.skywalking.network.grpc.provider;
|
|||
import com.a.eye.skywalking.network.grpc.AsyncTraceSearchServiceGrpc;
|
||||
import com.a.eye.skywalking.network.grpc.QueryTask;
|
||||
import com.a.eye.skywalking.network.grpc.SearchResult;
|
||||
import com.a.eye.skywalking.network.grpc.Span;
|
||||
import com.a.eye.skywalking.network.listener.AsyncTraceSearchListener;
|
||||
import io.grpc.stub.StreamObserver;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* Created by xin on 2016/11/15.
|
||||
*/
|
||||
public class AsyncTraceSearchService extends AsyncTraceSearchServiceGrpc.AsyncTraceSearchServiceImplBase {
|
||||
|
||||
public AsyncTraceSearchService(AsyncTraceSearchListener asyncTraceSearchListener) {
|
||||
private AsyncTraceSearchListener searchListener;
|
||||
|
||||
public AsyncTraceSearchService(AsyncTraceSearchListener searchListener) {
|
||||
this.searchListener = searchListener;
|
||||
}
|
||||
|
||||
@Override
|
||||
public StreamObserver<QueryTask> search(StreamObserver<SearchResult> responseObserver) {
|
||||
return super.search(responseObserver);
|
||||
public StreamObserver<QueryTask> search(final StreamObserver<SearchResult> responseObserver) {
|
||||
return new StreamObserver<QueryTask>() {
|
||||
private List<Span> spans;
|
||||
private int taskId;
|
||||
|
||||
@Override
|
||||
public void onNext(QueryTask value) {
|
||||
taskId = value.getTaskId();
|
||||
spans = searchListener.search(value.getTraceId());
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onError(Throwable t) {
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onCompleted() {
|
||||
responseObserver.onNext(SearchResult.newBuilder().addAllSpans(spans).setTaskId(taskId).build());
|
||||
responseObserver.onCompleted();
|
||||
}
|
||||
};
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,7 +1,13 @@
|
|||
package com.a.eye.skywalking.network.listener;
|
||||
|
||||
import com.a.eye.skywalking.network.grpc.Span;
|
||||
import com.a.eye.skywalking.network.grpc.TraceId;
|
||||
|
||||
import java.util.List;
|
||||
|
||||
/**
|
||||
* Created by xin on 2016/11/15.
|
||||
*/
|
||||
public interface AsyncTraceSearchListener {
|
||||
List<Span> search(TraceId traceId);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -21,6 +21,6 @@ message QueryTask {
|
|||
}
|
||||
|
||||
message SearchResult {
|
||||
string traceid = 1;
|
||||
int32 taskId = 1;
|
||||
repeated Span spans = 2;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -51,7 +51,7 @@ public class Main {
|
|||
IndexDataCapacityMonitor.start();
|
||||
|
||||
provider = ServiceProvider.newBuilder(Config.Server.PORT).addSpanStorageService(new StorageListener())
|
||||
.addTraceSearchService(new SearchListener()).build();
|
||||
.addAsyncTraceSearchService(new SearchListener()).build();
|
||||
provider.start();
|
||||
|
||||
if (logger.isDebugEnable()) {
|
||||
|
|
@ -63,6 +63,7 @@ public class Main {
|
|||
logger.info("SkyWalking storage server started.");
|
||||
Thread.currentThread().join();
|
||||
} catch (Throwable e) {
|
||||
e.printStackTrace();
|
||||
logger.error("SkyWalking storage server start failure.", e);
|
||||
} finally {
|
||||
provider.stop();
|
||||
|
|
|
|||
|
|
@ -17,7 +17,8 @@ public class Constants {
|
|||
+ " tid_s4 INT NOT NULL,\n"
|
||||
+ " tid_s5 INT NOT NULL,\n"
|
||||
+ " span_type INT NOT NULL, \n"
|
||||
+ " file_name VARCHAR(32) NOT NULL,\n"
|
||||
+ " file_name BIGINT NOT NULL,\n"
|
||||
+ " file_name_suffix INT NOT NULL,\n"
|
||||
+ " offset BIGINT NOT NULL,\n"
|
||||
+ " length INT NOT NULL\n" + ");\n";
|
||||
|
||||
|
|
@ -25,14 +26,14 @@ public class Constants {
|
|||
+ "(tid_s0,tid_s1,tid_s2,tid_s3,tid_s4,tid_s5);";
|
||||
|
||||
public static final String INSERT_INDEX = "INSERT INTO " +TABLE_NAME + "(tid_s0,tid_s1,tid_s2,tid_s3,tid_s4,tid_s5,span_type"
|
||||
+ ",file_name,offset,length) VALUES(?,?,?,?,?,?,?,?,?,?)";
|
||||
+ ",file_name,file_name_suffix,offset,length) VALUES(?,?,?,?,?,?,?,?,?,?,?)";
|
||||
|
||||
public static final String QUERY_TABLES = "SELECT count(1) AS TABLE_COUNT FROM INFORMATION_SCHEMA.TABLES "
|
||||
+ "WHERE TABLE_NAME= '" + TABLE_NAME.toUpperCase() + "';";
|
||||
|
||||
public static final String QUERY_INDEX_SIZE = "SELECT count(1) AS INDEX_SIZE FROM " + TABLE_NAME;
|
||||
|
||||
public static final String QUERY_TRACE_ID = "SELECT span_type, file_name, offset, length "
|
||||
public static final String QUERY_TRACE_ID = "SELECT span_type, file_name,file_name_suffix, offset, length "
|
||||
+ " FROM "+ TABLE_NAME+ " WHERE tid_s0 = ? AND tid_s1 = ? AND tid_s2 = ? AND tid_s3=? AND tid_s4=? AND"
|
||||
+ " tid_s5 = ?";
|
||||
|
||||
|
|
|
|||
|
|
@ -2,6 +2,7 @@ package com.a.eye.skywalking.storage.data;
|
|||
|
||||
import com.a.eye.skywalking.logging.api.ILog;
|
||||
import com.a.eye.skywalking.logging.api.LogManager;
|
||||
import com.a.eye.skywalking.network.grpc.TraceId;
|
||||
import com.a.eye.skywalking.storage.block.index.BlockIndexEngine;
|
||||
import com.a.eye.skywalking.storage.config.Config;
|
||||
import com.a.eye.skywalking.storage.config.Constants;
|
||||
|
|
@ -21,12 +22,11 @@ import static com.a.eye.skywalking.storage.config.Constants.SQL.DEFAULT_USER;
|
|||
import static com.a.eye.skywalking.storage.util.PathResolver.getAbsolutePath;
|
||||
|
||||
public class SpanDataFinder {
|
||||
private static ILog logger = LogManager.getLogger(SpanDataFinder.class);
|
||||
private static IndexDataSourceCache datasourceCache = new IndexDataSourceCache(Config.Finder.CACHED_SIZE);
|
||||
private static ReentrantLock createDatasourceLock = new ReentrantLock();
|
||||
|
||||
public static List<SpanData> find(String traceId) {
|
||||
long blockIndex = BlockIndexEngine.newFinder().find(fetchStartTimeFromTraceId(traceId));
|
||||
public static List<SpanData> find(TraceId traceId) {
|
||||
long blockIndex = BlockIndexEngine.newFinder().find(traceId.getSegments(1));
|
||||
if (blockIndex == 0) {
|
||||
return new ArrayList<SpanData>();
|
||||
}
|
||||
|
|
@ -35,7 +35,8 @@ public class SpanDataFinder {
|
|||
IndexMetaCollection indexMetaCollection = null;
|
||||
try {
|
||||
indexDBConnector = fetchIndexDBConnector(blockIndex);
|
||||
indexMetaCollection = indexDBConnector.queryByTraceId(spiltTraceId(traceId));
|
||||
indexMetaCollection = indexDBConnector.queryByTraceId(traceId.getSegmentsList().toArray(new Long[traceId
|
||||
.getSegmentsCount()]));
|
||||
} finally {
|
||||
if (indexDBConnector != null) {
|
||||
indexDBConnector.close();
|
||||
|
|
@ -49,7 +50,7 @@ public class SpanDataFinder {
|
|||
Iterator<IndexMetaGroup<String>> iterator = IndexMetaCollections.group(indexMetaCollection, new GroupKeyBuilder<String>() {
|
||||
@Override
|
||||
public String buildKey(IndexMetaInfo metaInfo) {
|
||||
return metaInfo.getFileName();
|
||||
return metaInfo.getFileName().fileName();
|
||||
}
|
||||
}).iterator();
|
||||
|
||||
|
|
|
|||
|
|
@ -1,6 +1,5 @@
|
|||
package com.a.eye.skywalking.storage.data.file;
|
||||
|
||||
import com.a.eye.datacarrier.common.AtomicRangeInteger;
|
||||
import com.a.eye.skywalking.health.report.HealthCollector;
|
||||
import com.a.eye.skywalking.health.report.HeathReading;
|
||||
import com.a.eye.skywalking.logging.api.ILog;
|
||||
|
|
@ -16,9 +15,6 @@ import java.io.File;
|
|||
import java.io.FileInputStream;
|
||||
import java.io.FileOutputStream;
|
||||
import java.io.IOException;
|
||||
import java.text.SimpleDateFormat;
|
||||
import java.util.Date;
|
||||
import java.util.concurrent.atomic.AtomicInteger;
|
||||
|
||||
import static com.a.eye.skywalking.storage.util.PathResolver.getAbsolutePath;
|
||||
|
||||
|
|
@ -28,10 +24,10 @@ import static com.a.eye.skywalking.storage.util.PathResolver.getAbsolutePath;
|
|||
public class DataFile {
|
||||
|
||||
private static ILog logger = LogManager.getLogger(DataFile.class);
|
||||
private String fileName;
|
||||
private DataFileNameDesc nameDesc;
|
||||
private long currentOffset;
|
||||
private DataFileOperator operator;
|
||||
private static final AtomicRangeInteger DATA_FILE_NAME_SUFFIX = new AtomicRangeInteger(1000, 9999);
|
||||
|
||||
|
||||
static {
|
||||
File dataFileDir = new File(getAbsolutePath(Config.DataFile.PATH));
|
||||
|
|
@ -41,21 +37,27 @@ public class DataFile {
|
|||
}
|
||||
|
||||
public DataFile() {
|
||||
this.fileName = new SimpleDateFormat("yyyy_MM_dd_HH_mm_ss_SS").format(new Date()) + "_" + DATA_FILE_NAME_SUFFIX
|
||||
.getAndIncrement();
|
||||
this.nameDesc = new DataFileNameDesc();
|
||||
this.currentOffset = 0;
|
||||
operator = new DataFileOperator();
|
||||
createFile();
|
||||
}
|
||||
|
||||
public DataFile(String fileName) {
|
||||
this.fileName = fileName;
|
||||
public DataFile(String fileName){
|
||||
this.nameDesc = new DataFileNameDesc(fileName);
|
||||
this.currentOffset = 0;
|
||||
operator = new DataFileOperator();
|
||||
createFile();
|
||||
}
|
||||
|
||||
public DataFile(DataFileNameDesc nameDesc) {
|
||||
this.nameDesc = nameDesc;
|
||||
operator = new DataFileOperator();
|
||||
createFile();
|
||||
}
|
||||
|
||||
public DataFile(File file) {
|
||||
this.fileName = file.getName();
|
||||
this.nameDesc = new DataFileNameDesc(file.getName());
|
||||
this.currentOffset = file.length();
|
||||
operator = new DataFileOperator();
|
||||
createFile();
|
||||
|
|
@ -67,7 +69,7 @@ public class DataFile {
|
|||
try {
|
||||
dataFile.createNewFile();
|
||||
if (logger.isDebugEnable()) {
|
||||
logger.debug("Create an new data file[{}].", fileName);
|
||||
logger.debug("Create an new data file[{}].", nameDesc.fileName());
|
||||
}
|
||||
HealthCollector.getCurrentHeathReading("DataFile")
|
||||
.updateData(HeathReading.INFO, "Create an new data " + "file.");
|
||||
|
|
@ -81,7 +83,7 @@ public class DataFile {
|
|||
public boolean overLimitLength() {
|
||||
boolean isOverLimitLength = currentOffset >= Config.DataFile.SIZE;
|
||||
if (isOverLimitLength) {
|
||||
logger.info("Data File[{}] is over limit length.", fileName);
|
||||
logger.info("Data File[{}] is over limit length.", nameDesc.fileName());
|
||||
}
|
||||
return isOverLimitLength;
|
||||
}
|
||||
|
|
@ -90,7 +92,7 @@ public class DataFile {
|
|||
byte[] bytes = data.toByteArray();
|
||||
try {
|
||||
operator.getWriter().write(bytes);
|
||||
IndexMetaInfo metaInfo = new IndexMetaInfo(data, fileName, currentOffset, bytes.length);
|
||||
IndexMetaInfo metaInfo = new IndexMetaInfo(data, nameDesc, currentOffset, bytes.length);
|
||||
currentOffset += bytes.length;
|
||||
return metaInfo;
|
||||
} catch (IOException e) {
|
||||
|
|
@ -118,7 +120,7 @@ public class DataFile {
|
|||
return data;
|
||||
} catch (IOException e) {
|
||||
throw new SpanDataReadFailedException(
|
||||
"Failed to read dataFile[" + fileName + "], offset: " + offset + " " + "lenght: " + length, e);
|
||||
"Failed to read dataFile[" + nameDesc.fileName() + "], offset: " + offset + " " + "lenght: " + length, e);
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -169,12 +171,13 @@ public class DataFile {
|
|||
}
|
||||
}
|
||||
|
||||
|
||||
private File getDataFile() {
|
||||
return new File(getAbsolutePath(Config.DataFile.PATH), fileName);
|
||||
return new File(getAbsolutePath(Config.DataFile.PATH), nameDesc.fileName());
|
||||
}
|
||||
|
||||
@Override
|
||||
public String toString() {
|
||||
return "DataFile{" + "fileName='" + fileName + '\'' + '}';
|
||||
return "DataFile{" + "fileName='" + nameDesc.fileName() + '\'' + '}';
|
||||
}
|
||||
}
|
||||
|
|
|
|||
|
|
@ -0,0 +1,74 @@
|
|||
package com.a.eye.skywalking.storage.data.file;
|
||||
|
||||
import com.a.eye.datacarrier.common.AtomicRangeInteger;
|
||||
|
||||
import java.text.ParseException;
|
||||
import java.text.SimpleDateFormat;
|
||||
|
||||
/**
|
||||
* Created by xin on 2016/11/16.
|
||||
*/
|
||||
public class DataFileNameDesc {
|
||||
private static final AtomicRangeInteger DATA_FILE_NAME_SUFFIX = new AtomicRangeInteger(1000, 9999);
|
||||
private long name;
|
||||
private int suffix;
|
||||
private String fileNameStr;
|
||||
|
||||
public DataFileNameDesc() {
|
||||
name = System.currentTimeMillis();
|
||||
suffix = DATA_FILE_NAME_SUFFIX.getAndIncrement();
|
||||
fileNameStr = new SimpleDateFormat("yyyy_MM_dd_HH_mm_ss_SS").format(name) + "_" + suffix;
|
||||
}
|
||||
|
||||
public DataFileNameDesc(long name, int suffix) {
|
||||
this.name = name;
|
||||
this.suffix = suffix;
|
||||
fileNameStr = new SimpleDateFormat("yyyy_MM_dd_HH_mm_ss_SS").format(name) + "_" + suffix;
|
||||
}
|
||||
|
||||
public DataFileNameDesc(String fileName) {
|
||||
int lastIndex = fileName.lastIndexOf('_');
|
||||
try {
|
||||
this.name = new SimpleDateFormat("yyyy_MM_dd_HH_mm_ss_SS").parse(fileName.substring(0, lastIndex - 1))
|
||||
.getTime();
|
||||
} catch (ParseException e) {
|
||||
}
|
||||
this.suffix = Integer.parseInt(fileName.substring(lastIndex + 1));
|
||||
fileNameStr = fileName;
|
||||
}
|
||||
|
||||
|
||||
public String fileName() {
|
||||
return fileNameStr;
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean equals(Object o) {
|
||||
if (this == o)
|
||||
return true;
|
||||
if (o == null || getClass() != o.getClass())
|
||||
return false;
|
||||
|
||||
DataFileNameDesc that = (DataFileNameDesc) o;
|
||||
|
||||
if (name != that.name)
|
||||
return false;
|
||||
return suffix == that.suffix;
|
||||
|
||||
}
|
||||
|
||||
@Override
|
||||
public int hashCode() {
|
||||
int result = (int) (name ^ (name >>> 32));
|
||||
result = 31 * result + suffix;
|
||||
return result;
|
||||
}
|
||||
|
||||
public long getName() {
|
||||
return name;
|
||||
}
|
||||
|
||||
public int getSuffix() {
|
||||
return suffix;
|
||||
}
|
||||
}
|
||||
|
|
@ -5,6 +5,7 @@ import com.a.eye.skywalking.logging.api.LogManager;
|
|||
import com.a.eye.skywalking.storage.config.Config;
|
||||
import com.a.eye.skywalking.storage.config.Constants;
|
||||
import com.a.eye.skywalking.storage.data.exception.ConnectorInitializeFailedException;
|
||||
import com.a.eye.skywalking.storage.data.file.DataFileNameDesc;
|
||||
import com.a.eye.skywalking.storage.data.spandata.AckSpanData;
|
||||
import com.a.eye.skywalking.storage.data.spandata.RequestSpanData;
|
||||
import com.a.eye.skywalking.storage.data.spandata.SpanData;
|
||||
|
|
@ -109,9 +110,10 @@ public class IndexDBConnector {
|
|||
ps.setInt(5, metaInfo.getTraceId()[4].intValue());
|
||||
ps.setInt(6, metaInfo.getTraceId()[5].intValue());
|
||||
ps.setInt(7, metaInfo.getSpanType().getValue());
|
||||
ps.setString(8, metaInfo.getFileName());
|
||||
ps.setLong(9, metaInfo.getOffset());
|
||||
ps.setInt(10, metaInfo.getLength());
|
||||
ps.setLong(8, metaInfo.getFileName().getName());
|
||||
ps.setInt(9, metaInfo.getFileName().getSuffix());
|
||||
ps.setLong(10, metaInfo.getOffset());
|
||||
ps.setInt(11, metaInfo.getLength());
|
||||
ps.addBatch();
|
||||
if (++currentIndex > Constants.MAX_BATCH_SIZE) {
|
||||
ps.executeBatch();
|
||||
|
|
@ -144,15 +146,15 @@ public class IndexDBConnector {
|
|||
return indexSize;
|
||||
}
|
||||
|
||||
public IndexMetaCollection queryByTraceId(long[] traceId) {
|
||||
public IndexMetaCollection queryByTraceId(Long[] traceId) {
|
||||
try {
|
||||
PreparedStatement ps = connection.prepareStatement(QUERY_TRACE_ID);
|
||||
ps.setInt(1, (int) traceId[0]);
|
||||
ps.setLong(2, (int) traceId[1]);
|
||||
ps.setInt(3, (int) traceId[2]);
|
||||
ps.setInt(4, (int) traceId[3]);
|
||||
ps.setInt(5, (int) traceId[4]);
|
||||
ps.setInt(6, (int) traceId[5]);
|
||||
ps.setInt(1, traceId[0].intValue());
|
||||
ps.setLong(2, traceId[1]);
|
||||
ps.setInt(3, traceId[2].intValue());
|
||||
ps.setInt(4, traceId[3].intValue());
|
||||
ps.setInt(5, traceId[4].intValue());
|
||||
ps.setInt(6, traceId[5].intValue());
|
||||
ResultSet rs = ps.executeQuery();
|
||||
|
||||
IndexMetaCollection collection = new IndexMetaCollection();
|
||||
|
|
@ -166,8 +168,9 @@ public class IndexDBConnector {
|
|||
spanData = new RequestSpanData();
|
||||
}
|
||||
|
||||
collection.add(new IndexMetaInfo(spanData, rs.getString("file_name"), rs.getLong("offset"),
|
||||
rs.getInt("length")));
|
||||
collection.add(new IndexMetaInfo(spanData,
|
||||
new DataFileNameDesc(rs.getLong("file_name"), rs.getInt("file_name_suffix")),
|
||||
rs.getLong("offset"), rs.getInt("length")));
|
||||
}
|
||||
return collection;
|
||||
} catch (SQLException e) {
|
||||
|
|
@ -183,4 +186,6 @@ public class IndexDBConnector {
|
|||
logger.error("Failed to close index db connector", e);
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,5 +1,6 @@
|
|||
package com.a.eye.skywalking.storage.data.index;
|
||||
|
||||
import com.a.eye.skywalking.storage.data.file.DataFileNameDesc;
|
||||
import com.a.eye.skywalking.storage.data.spandata.SpanData;
|
||||
import com.a.eye.skywalking.storage.data.spandata.SpanType;
|
||||
|
||||
|
|
@ -7,21 +8,21 @@ public class IndexMetaInfo {
|
|||
|
||||
private SpanData spanData;
|
||||
|
||||
private String fileName;
|
||||
private DataFileNameDesc nameDesc;
|
||||
|
||||
private long offset;
|
||||
|
||||
private int length;
|
||||
|
||||
public IndexMetaInfo(SpanData data, String fileName, long offset, int length) {
|
||||
public IndexMetaInfo(SpanData data, DataFileNameDesc fileNameDesc, long offset, int length) {
|
||||
this.spanData = data;
|
||||
this.fileName = fileName;
|
||||
this.nameDesc = fileNameDesc;
|
||||
this.offset = offset;
|
||||
this.length = length;
|
||||
}
|
||||
|
||||
public String getFileName() {
|
||||
return fileName;
|
||||
public DataFileNameDesc getFileName() {
|
||||
return nameDesc;
|
||||
}
|
||||
|
||||
public long getOffset() {
|
||||
|
|
|
|||
|
|
@ -5,6 +5,8 @@ import com.a.eye.skywalking.health.report.HeathReading;
|
|||
import com.a.eye.skywalking.logging.api.ILog;
|
||||
import com.a.eye.skywalking.logging.api.LogManager;
|
||||
import com.a.eye.skywalking.network.grpc.Span;
|
||||
import com.a.eye.skywalking.network.grpc.TraceId;
|
||||
import com.a.eye.skywalking.network.listener.AsyncTraceSearchListener;
|
||||
import com.a.eye.skywalking.network.listener.TraceSearchListener;
|
||||
import com.a.eye.skywalking.storage.data.SpanDataFinder;
|
||||
import com.a.eye.skywalking.storage.data.spandata.SpanData;
|
||||
|
|
@ -13,12 +15,12 @@ import com.a.eye.skywalking.storage.data.spandata.SpanDataHelper;
|
|||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
|
||||
public class SearchListener implements TraceSearchListener {
|
||||
public class SearchListener implements AsyncTraceSearchListener {
|
||||
|
||||
private static ILog logger = LogManager.getLogger(SearchListener.class);
|
||||
|
||||
@Override
|
||||
public List<Span> search(String traceId) {
|
||||
public List<Span> search(TraceId traceId) {
|
||||
try {
|
||||
List<SpanData> data = SpanDataFinder.find(traceId);
|
||||
SpanDataHelper helper = new SpanDataHelper(data);
|
||||
|
|
|
|||
|
|
@ -0,0 +1,46 @@
|
|||
package com.a.eye.skywalking.storage;
|
||||
|
||||
import com.a.eye.skywalking.network.dependencies.io.grpc.ManagedChannel;
|
||||
import com.a.eye.skywalking.network.dependencies.io.grpc.ManagedChannelBuilder;
|
||||
import com.a.eye.skywalking.network.dependencies.io.grpc.stub.StreamObserver;
|
||||
import com.a.eye.skywalking.network.grpc.*;
|
||||
|
||||
import static com.a.eye.skywalking.network.grpc.AsyncTraceSearchServiceGrpc.newStub;
|
||||
|
||||
|
||||
public class SearchClient {
|
||||
private static ManagedChannel channel =
|
||||
ManagedChannelBuilder.forAddress("127.0.0.1", 34000).usePlaintext(true).build();
|
||||
|
||||
private static AsyncTraceSearchServiceGrpc.AsyncTraceSearchServiceStub searchServiceStub = newStub(channel);
|
||||
|
||||
|
||||
|
||||
public static void main(String[] args) throws InterruptedException {
|
||||
StreamObserver<SearchResult> serverStreamObserver = new StreamObserver<SearchResult>() {
|
||||
@Override
|
||||
public void onNext(SearchResult searchResult) {
|
||||
System.out.println(searchResult.getSpansCount());
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onError(Throwable throwable) {
|
||||
|
||||
}
|
||||
|
||||
@Override
|
||||
public void onCompleted() {
|
||||
|
||||
}
|
||||
};
|
||||
StreamObserver<QueryTask> searchResult = searchServiceStub.search(serverStreamObserver);
|
||||
|
||||
searchResult.onNext(QueryTask.newBuilder().setTraceId(
|
||||
TraceId.newBuilder().addSegments(201611).addSegments(1478661327960L).addSegments(8504828)
|
||||
.addSegments(2277).addSegments(53).addSegments(3).build()).setTaskId(1).build());
|
||||
searchResult.onCompleted();
|
||||
|
||||
Thread.sleep(10000);
|
||||
|
||||
}
|
||||
}
|
||||
Loading…
Reference in New Issue