x00384074 commented on issue #2417:
URL:
https://github.com/apache/servicecomb-java-chassis/issues/2417#issuecomment-862148463
实现这个会影响吗
public class PumpImpl<T> implements Pump {
private final ReadStream<T> readStream;
private final WriteStream<T> writeStream;
private final Handler<T> dataHandler;
private final Handler<Void> drainHandler;
private int pumped;
/**
* Create a new {@code Pump} with the given {@code ReadStream} and
{@code WriteStream}. Set the write queue max size
* of the write stream to {@code maxWriteQueueSize}
*/
public PumpImpl(ReadStream<T> rs, WriteStream<T> ws, int
maxWriteQueueSize) {
this(rs, ws);
writeStream.setWriteQueueMaxSize(maxWriteQueueSize);
}
public PumpImpl(ReadStream<T> rs, WriteStream<T> ws) {
readStream = rs;
writeStream = ws;
drainHandler = v -> readStream.resume();
writeStream.drainHandler(drainHandler);
dataHandler = data -> {
writeStream.write(data);
incPumped();
if (writeStream.writeQueueFull()) {
readStream.pause();
}
};
}
/**
* 设置写队列最大长度
*
* @param maxSize 队列最大长度
* @return PumpImpl
*/
@Override
public PumpImpl setWriteQueueMaxSize(int maxSize) {
writeStream.setWriteQueueMaxSize(maxSize);
return this;
}
/**
* 开始处理
*
* @return PumpImpl
*/
@Override
public PumpImpl start() {
readStream.handler(dataHandler);
return this;
}
/**
* 停止
*
* @return PumpImpl
*/
@Override
public PumpImpl stop() {
writeStream.drainHandler(null);
readStream.handler(null);
return this;
}
/**
* numberPumped
*
* @return int
*/
@Override
public synchronized int numberPumped() {
return pumped;
}
// Note we synchronize as numberPumped can be called from a different
thread however incPumped will always
// be called from the same thread so we benefit from bias locked
optimisation which should give a very low
// overhead
private synchronized void incPumped() {
pumped++;
}
}
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
