kazusa
/
skywalking-java
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

1. change grpc facade, to open more interface to client.

This commit is contained in:
wusheng 2016-11-27 11:29:36 +08:00
parent dcda4e766f
commit c99679a868
15 changed files with 87 additions and 68 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
skywalking-network/src/main/java/com/a/eye/skywalking/network/Client.java
View File

@ -1,30 +1,24 @@
package com.a.eye.skywalking.network;
import com.a.eye.skywalking.network.grpc.SpanStorageServiceGrpc;
import com.a.eye.skywalking.network.grpc.TraceSearchServiceGrpc;
import com.a.eye.skywalking.network.grpc.client.SpanStorageClient;
import com.a.eye.skywalking.network.grpc.client.TraceSearchClient;
import com.a.eye.skywalking.network.listener.client.StorageClientListener;
import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
public class Client {
private ManagedChannel channel;
private SpanStorageServiceGrpc.SpanStorageServiceStub spanStorageStub;
private TraceSearchServiceGrpc.TraceSearchServiceStub traceSearchServiceStub;
public Client(String ip, int address) {
channel = ManagedChannelBuilder.forAddress(ip, address).usePlaintext(true).build();
spanStorageStub = SpanStorageServiceGrpc.newStub(channel);
traceSearchServiceStub = TraceSearchServiceGrpc.newStub(channel);
}
public SpanStorageClient newSpanStorageClient(StorageClientListener listener) {
return new SpanStorageClient(channel, listener);
}
public SpanStorageClient newSpanStorageClient() {
return new SpanStorageClient(spanStorageStub);
}
public TraceSearchClient newTraceSearchClient(){
return new TraceSearchClient(traceSearchServiceStub);
public TraceSearchClient newTraceSearchClient(StorageClientListener listener){
return new TraceSearchClient(channel, listener);
}
}

14
skywalking-network/src/main/java/com/a/eye/skywalking/network/Server.java
View File

@ -3,9 +3,9 @@ package com.a.eye.skywalking.network;
import com.a.eye.skywalking.network.grpc.server.AsyncTraceSearchServer;
import com.a.eye.skywalking.network.grpc.server.SpanStorageServer;
import com.a.eye.skywalking.network.grpc.server.TraceSearchServer;
import com.a.eye.skywalking.network.listener.AsyncTraceSearchListener;
import com.a.eye.skywalking.network.listener.SpanStorageListener;
import com.a.eye.skywalking.network.listener.TraceSearchListener;
import com.a.eye.skywalking.network.listener.server.AsyncTraceSearchServerListener;
import com.a.eye.skywalking.network.listener.server.SpanStorageServerListener;
import com.a.eye.skywalking.network.listener.server.TraceSearchListener;
import io.grpc.netty.NettyServerBuilder;
import io.netty.channel.nio.NioEventLoopGroup;
@ -51,8 +51,8 @@ public class Server {
.workerEventLoopGroup(new NioEventLoopGroup()).build());
}
public TransferServiceBuilder addSpanStorageService(SpanStorageListener spanStorageListener) {
serverBuilder.addService(new SpanStorageServer(spanStorageListener));
public TransferServiceBuilder addSpanStorageService(SpanStorageServerListener spanStorageServerListener) {
serverBuilder.addService(new SpanStorageServer(spanStorageServerListener));
return this;
}
@ -61,8 +61,8 @@ public class Server {
return this;
}
public TransferServiceBuilder addAsyncTraceSearchService(AsyncTraceSearchListener asyncTraceSearchListener){
serverBuilder.addService(new AsyncTraceSearchServer(asyncTraceSearchListener));
public TransferServiceBuilder addAsyncTraceSearchService(AsyncTraceSearchServerListener asyncTraceSearchServerListener){
serverBuilder.addService(new AsyncTraceSearchServer(asyncTraceSearchServerListener));
return this;
}
}

9
skywalking-network/src/main/java/com/a/eye/skywalking/network/exception/ConsumeSpanDataFailedException.java
View File

@ -1,9 +0,0 @@
package com.a.eye.skywalking.network.exception;
/**
* Created by xin on 2016/11/23.
*/
public class ConsumeSpanDataFailedException extends RuntimeException {
public ConsumeSpanDataFailedException(Exception e) {
}
}

30
skywalking-network/src/main/java/com/a/eye/skywalking/network/grpc/client/SpanStorageClient.java
View File

@ -1,10 +1,11 @@
package com.a.eye.skywalking.network.grpc.client;
import com.a.eye.skywalking.network.exception.ConsumeSpanDataFailedException;
import com.a.eye.skywalking.network.grpc.AckSpan;
import com.a.eye.skywalking.network.grpc.RequestSpan;
import com.a.eye.skywalking.network.grpc.SendResult;
import com.a.eye.skywalking.network.grpc.SpanStorageServiceGrpc;
import com.a.eye.skywalking.network.listener.client.StorageClientListener;
import io.grpc.ManagedChannel;
import io.grpc.stub.CallStreamObserver;
import io.grpc.stub.StreamObserver;
@ -12,8 +13,11 @@ public class SpanStorageClient {
private final SpanStorageServiceGrpc.SpanStorageServiceStub spanStorageStub;
public SpanStorageClient(SpanStorageServiceGrpc.SpanStorageServiceStub spanStorageStub) {
this.spanStorageStub = spanStorageStub;
private final StorageClientListener listener;
public SpanStorageClient(ManagedChannel channel, StorageClientListener listener) {
this.spanStorageStub = SpanStorageServiceGrpc.newStub(channel);
this.listener = listener;
}
public void sendRequestSpan(RequestSpan... requestSpan) {
@ -21,11 +25,12 @@ public class SpanStorageClient {
spanStorageStub.storageRequestSpan(new StreamObserver<SendResult>() {
@Override
public void onNext(SendResult sendResult) {
listener.onBatchFinished(sendResult);
}
@Override
public void onError(Throwable throwable) {
throwable.printStackTrace();
listener.onError(throwable);
}
@Override
@ -37,13 +42,6 @@ public class SpanStorageClient {
for (RequestSpan span : requestSpan) {
requestSpanStreamObserver.onNext(span);
}
while (!((CallStreamObserver<RequestSpan>) requestSpanStreamObserver).isReady()) {
try {
Thread.sleep(1);
} catch (InterruptedException e) {
throw new ConsumeSpanDataFailedException(e);
}
}
requestSpanStreamObserver.onCompleted();
}
@ -53,11 +51,12 @@ public class SpanStorageClient {
spanStorageStub.storageACKSpan(new StreamObserver<SendResult>() {
@Override
public void onNext(SendResult sendResult) {
listener.onBatchFinished(sendResult);
}
@Override
public void onError(Throwable throwable) {
throwable.printStackTrace();
listener.onError(throwable);
}
@Override
@ -69,13 +68,6 @@ public class SpanStorageClient {
for (AckSpan span : ackSpan) {
ackSpanStreamObserver.onNext(span);
}
while (!((CallStreamObserver<AckSpan>) ackSpanStreamObserver).isReady()) {
try {
Thread.sleep(1);
} catch (InterruptedException e) {
throw new ConsumeSpanDataFailedException(e);
}
}
ackSpanStreamObserver.onCompleted();
}

6
skywalking-network/src/main/java/com/a/eye/skywalking/network/grpc/client/TraceSearchClient.java
View File

@ -1,6 +1,8 @@
package com.a.eye.skywalking.network.grpc.client;
import com.a.eye.skywalking.network.grpc.TraceSearchServiceGrpc;
import com.a.eye.skywalking.network.listener.client.StorageClientListener;
import io.grpc.ManagedChannel;
/**
* Created by wusheng on 2016/11/26.
@ -9,8 +11,8 @@ public class TraceSearchClient {
private final TraceSearchServiceGrpc.TraceSearchServiceStub traceSearchServiceStub;
public TraceSearchClient(TraceSearchServiceGrpc.TraceSearchServiceStub traceSearchServiceStub) {
this.traceSearchServiceStub = traceSearchServiceStub;
public TraceSearchClient(ManagedChannel channel, StorageClientListener listener) {
this.traceSearchServiceStub = TraceSearchServiceGrpc.newStub(channel);
}
}

6
skywalking-network/src/main/java/com/a/eye/skywalking/network/grpc/server/AsyncTraceSearchServer.java
View File

@ -4,7 +4,7 @@ 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 com.a.eye.skywalking.network.listener.server.AsyncTraceSearchServerListener;
import io.grpc.stub.StreamObserver;
import java.util.List;
@ -14,9 +14,9 @@ import java.util.List;
*/
public class AsyncTraceSearchServer extends AsyncTraceSearchServiceGrpc.AsyncTraceSearchServiceImplBase {
private AsyncTraceSearchListener searchListener;
private AsyncTraceSearchServerListener searchListener;
public AsyncTraceSearchServer(AsyncTraceSearchListener searchListener) {
public AsyncTraceSearchServer(AsyncTraceSearchServerListener searchListener) {
this.searchListener = searchListener;
}

6
skywalking-network/src/main/java/com/a/eye/skywalking/network/grpc/server/SpanStorageServer.java
View File

@ -4,13 +4,13 @@ import com.a.eye.skywalking.network.grpc.AckSpan;
import com.a.eye.skywalking.network.grpc.RequestSpan;
import com.a.eye.skywalking.network.grpc.SendResult;
import com.a.eye.skywalking.network.grpc.SpanStorageServiceGrpc;
import com.a.eye.skywalking.network.listener.SpanStorageListener;
import com.a.eye.skywalking.network.listener.server.SpanStorageServerListener;
import io.grpc.stub.StreamObserver;
public class SpanStorageServer extends SpanStorageServiceGrpc.SpanStorageServiceImplBase {
private SpanStorageListener listener;
public SpanStorageServer(SpanStorageListener listener) {
private SpanStorageServerListener listener;
public SpanStorageServer(SpanStorageServerListener listener) {
this.listener = listener;
}

2
skywalking-network/src/main/java/com/a/eye/skywalking/network/grpc/server/TraceSearchServer.java
View File

@ -3,7 +3,7 @@ package com.a.eye.skywalking.network.grpc.server;
import com.a.eye.skywalking.network.grpc.SearchResult;
import com.a.eye.skywalking.network.grpc.TraceId;
import com.a.eye.skywalking.network.grpc.TraceSearchServiceGrpc;
import com.a.eye.skywalking.network.listener.TraceSearchListener;
import com.a.eye.skywalking.network.listener.server.TraceSearchListener;
import io.grpc.stub.StreamObserver;
/**

12
skywalking-network/src/main/java/com/a/eye/skywalking/network/listener/client/StorageClientListener.java Normal file
View File

@ -0,0 +1,12 @@
package com.a.eye.skywalking.network.listener.client;
import com.a.eye.skywalking.network.grpc.SendResult;
/**
* Created by wusheng on 2016/11/27.
*/
public interface StorageClientListener {
void onError(Throwable throwable);
void onBatchFinished(SendResult sendResult);
}

4
skywalking-network/src/main/java/com/a/eye/skywalking/network/listener/AsyncTraceSearchListener.java → skywalking-network/src/main/java/com/a/eye/skywalking/network/listener/server/AsyncTraceSearchServerListener.java
View File

@ -1,4 +1,4 @@
package com.a.eye.skywalking.network.listener;
package com.a.eye.skywalking.network.listener.server;
import com.a.eye.skywalking.network.grpc.Span;
import com.a.eye.skywalking.network.grpc.TraceId;
@ -8,6 +8,6 @@ import java.util.List;
/**
* Created by xin on 2016/11/15.
*/
public interface AsyncTraceSearchListener {
public interface AsyncTraceSearchServerListener {
List<Span> search(TraceId traceId);
}

4
skywalking-network/src/main/java/com/a/eye/skywalking/network/listener/SpanStorageListener.java → skywalking-network/src/main/java/com/a/eye/skywalking/network/listener/server/SpanStorageServerListener.java
View File

@ -1,9 +1,9 @@
package com.a.eye.skywalking.network.listener;
package com.a.eye.skywalking.network.listener.server;
import com.a.eye.skywalking.network.grpc.AckSpan;
import com.a.eye.skywalking.network.grpc.RequestSpan;
public interface SpanStorageListener{
public interface SpanStorageServerListener {
boolean storage(RequestSpan requestSpan);
boolean storage(AckSpan ackSpan);

2
skywalking-network/src/main/java/com/a/eye/skywalking/network/listener/TraceSearchListener.java → skywalking-network/src/main/java/com/a/eye/skywalking/network/listener/server/TraceSearchListener.java
View File

@ -1,4 +1,4 @@
package com.a.eye.skywalking.network.listener;
package com.a.eye.skywalking.network.listener.server;
import com.a.eye.skywalking.network.grpc.Span;

5
skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/listener/SearchListener.java
View File

@ -6,8 +6,7 @@ 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.network.listener.server.AsyncTraceSearchServerListener;
import com.a.eye.skywalking.storage.data.SpanDataFinder;
import com.a.eye.skywalking.storage.data.spandata.SpanData;
import com.a.eye.skywalking.storage.data.spandata.SpanDataHelper;
@ -15,7 +14,7 @@ import com.a.eye.skywalking.storage.data.spandata.SpanDataHelper;
import java.util.ArrayList;
import java.util.List;
public class SearchListener implements AsyncTraceSearchListener {
public class SearchListener implements AsyncTraceSearchServerListener {
private static ILog logger = LogManager.getLogger(SearchListener.class);

4
skywalking-storage-center/skywalking-storage/src/main/java/com/a/eye/skywalking/storage/listener/StorageListener.java
View File

@ -6,7 +6,7 @@ import com.a.eye.skywalking.logging.api.ILog;
import com.a.eye.skywalking.logging.api.LogManager;
import com.a.eye.skywalking.network.grpc.AckSpan;
import com.a.eye.skywalking.network.grpc.RequestSpan;
import com.a.eye.skywalking.network.listener.SpanStorageListener;
import com.a.eye.skywalking.network.listener.server.SpanStorageServerListener;
import com.a.eye.skywalking.storage.config.Config;
import com.a.eye.skywalking.storage.data.spandata.AckSpanData;
import com.a.eye.skywalking.storage.data.spandata.RequestSpanData;
@ -18,7 +18,7 @@ import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.util.DaemonThreadFactory;
public class StorageListener implements SpanStorageListener {
public class StorageListener implements SpanStorageServerListener {
private ILog logger = LogManager.getLogger(StorageListener.class);

31
skywalking-storage-center/skywalking-storage/src/test/java/StorageThread.java
View File

@ -1,22 +1,28 @@
import com.a.eye.skywalking.network.Client;
import com.a.eye.skywalking.network.grpc.AckSpan;
import com.a.eye.skywalking.network.grpc.RequestSpan;
import com.a.eye.skywalking.network.grpc.SendResult;
import com.a.eye.skywalking.network.grpc.TraceId;
import com.a.eye.skywalking.network.grpc.client.SpanStorageClient;
import com.a.eye.skywalking.network.listener.client.StorageClientListener;
import com.a.eye.skywalking.storage.util.NetUtils;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.locks.LockSupport;
public class StorageThread extends Thread {
private SpanStorageClient client;
private long count;
private CountDownLatch countDownLatch;
private MyStorageClientListener listener;
StorageThread(long count, CountDownLatch countDownLatch) {
client = new Client("10.128.7.241", 34000).newSpanStorageClient();
listener = new MyStorageClientListener();
client = new Client("10.128.7.241", 34000).newSpanStorageClient(listener);
this.count = count;
this.countDownLatch = countDownLatch;
}
@ -42,6 +48,11 @@ public class StorageThread extends Thread {
client.sendACKSpan(ackSpanList);
client.sendRequestSpan(requestSpanList);
cycle = 0;
while(!listener.isCompleted){
LockSupport.parkNanos(1);
}
listener.begin();
}
requestSpanList[cycle] = requestSpan;
@ -55,4 +66,22 @@ public class StorageThread extends Thread {
countDownLatch.countDown();
}
public class MyStorageClientListener implements StorageClientListener{
volatile boolean isCompleted = false;
@Override
public void onError(Throwable throwable) {
}
@Override
public void onBatchFinished(SendResult sendResult) {
isCompleted = true;
}
public void begin(){
isCompleted = false;
}
}
}