/*
* Copyright (c) 2016, Oracle and/or its affiliates. All rights reserved.
* DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.
*
* This code is free software; you can redistribute it and/or modify it
* under the terms of the GNU General Public License version 2 only, as
* published by the Free Software Foundation. Oracle designates this
* particular file as subject to the "Classpath" exception as provided
* by Oracle in the LICENSE file that accompanied this code.
*
* This code is distributed in the hope that it will be useful, but WITHOUT
* ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
* FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License
* version 2 for more details (a copy is included in the LICENSE file that
* accompanied this code).
*
* You should have received a copy of the GNU General Public License version
* 2 along with this work; if not, write to the Free Software Foundation,
* Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.
*
* Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA
* or visit www.oracle.com if you need additional information or have any
* questions.
*/
package java.net.http;
import java.io.IOException;
import java.io.UncheckedIOException;
import java.net.ProtocolException;
import java.net.http.WebSocket.Listener;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.util.Optional;
import java.util.concurrent.CompletionStage;
import java.util.concurrent.Executor;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;
import java.util.function.Supplier;
import static java.lang.System.Logger.Level.ERROR;
import static java.net.http.WSUtils.EMPTY_BYTE_BUFFER;
import static java.net.http.WSUtils.logger;
/*
* Receives incoming data from the channel and converts it into a sequence of
* messages, which are then passed to the listener.
*/
final class WSReceiver {
private final Listener listener;
private final WebSocket webSocket;
private final Supplier<WSShared<ByteBuffer>> buffersSupplier =
new WSSharedPool<>(() -> ByteBuffer.allocateDirect(32768), 2);
private final RawChannel channel;
private final RawChannel.NonBlockingEvent channelEvent;
private final WSSignalHandler handler;
private final AtomicLong demand = new AtomicLong();
private final AtomicBoolean readable = new AtomicBoolean();
private boolean started;
private volatile boolean closed;
private final WSFrame.Reader reader = new WSFrame.Reader();
private final WSFrameConsumer frameConsumer;
private WSShared<ByteBuffer> buf = WSShared.wrap(EMPTY_BYTE_BUFFER);
private WSShared<ByteBuffer> data; // TODO: initialize with leftovers from the RawChannel
WSReceiver(Listener listener, WebSocket webSocket, Executor executor,
RawChannel channel) {
this.listener = listener;
this.webSocket = webSocket;
this.channel = channel;
handler = new WSSignalHandler(executor, this::react);
channelEvent = createChannelEvent();
this.frameConsumer = new WSFrameConsumer(new MessageConsumer());
}
private void react() {
synchronized (this) {
while (demand.get() > 0 && !closed) {
try {
if (data == null) {
if (!getData()) {
break;
}
}
reader.readFrame(data, frameConsumer);
if (!data.hasRemaining()) {
data.dispose();
data = null;
}
// In case of exception we don't need to clean any state,
// since it's the terminal condition anyway. Nothing will be
// retried.
} catch (WSProtocolException e) {
// Translate into ProtocolException
closeExceptionally(new ProtocolException().initCause(e));
} catch (Exception e) {
closeExceptionally(e);
}
}
}
}
long request(long n) {
long newDemand = demand.accumulateAndGet(n, (p, i) -> p + i < 0 ? Long.MAX_VALUE : p + i);
handler.signal();
assert newDemand >= 0 : newDemand;
return newDemand;
}
private boolean getData() throws IOException {
if (!readable.get()) {
return false;
}
if (!buf.hasRemaining()) {
buf.dispose();
buf = buffersSupplier.get();
assert buf.hasRemaining() : buf;
}
int oldPosition = buf.position();
int oldLimit = buf.limit();
int numRead = channel.read(buf.buffer());
if (numRead > 0) {
data = buf.share(oldPosition, oldPosition + numRead);
buf.select(buf.limit(), oldLimit); // Move window to the free region
return true;
} else if (numRead == 0) {
readable.set(false);
channel.registerEvent(channelEvent);
return false;
} else {
assert numRead < 0 : numRead;
throw new WSProtocolException
("7.2.1.", "Stream ended before a Close frame has been received");
}
}
void start() {
synchronized (this) {
if (started) {
throw new IllegalStateException("Already started");
}
started = true;
try {
channel.registerEvent(channelEvent);
} catch (IOException e) {
throw new UncheckedIOException(e);
}
try {
listener.onOpen(webSocket);
} catch (Exception e) {
closeExceptionally(new RuntimeException("onOpen threw an exception", e));
}
}
}
private void close() { // TODO: move to WS.java
closed = true;
}
private void closeExceptionally(Throwable error) { // TODO: move to WS.java
close();
try {
listener.onError(webSocket, error);
} catch (Exception e) {
logger.log(ERROR, "onError threw an exception", e);
}
}
private final class MessageConsumer implements WSMessageConsumer {
@Override
public void onText(WebSocket.MessagePart part, WSDisposableText data) {
decrementDemand();
CompletionStage<?> cs;
try {
cs = listener.onText(webSocket, data, part);
} catch (Exception e) {
closeExceptionally(new RuntimeException("onText threw an exception", e));
return;
}
follow(cs, data, "onText");
}
@Override
public void onBinary(WebSocket.MessagePart part, WSShared<ByteBuffer> data) {
decrementDemand();
CompletionStage<?> cs;
try {
cs = listener.onBinary(webSocket, data.buffer(), part);
} catch (Exception e) {
closeExceptionally(new RuntimeException("onBinary threw an exception", e));
return;
}
follow(cs, data, "onBinary");
}
@Override
public void onPing(WSShared<ByteBuffer> data) {
decrementDemand();
CompletionStage<?> cs;
try {
cs = listener.onPing(webSocket, data.buffer());
} catch (Exception e) {
closeExceptionally(new RuntimeException("onPing threw an exception", e));
return;
}
follow(cs, data, "onPing");
}
@Override
public void onPong(WSShared<ByteBuffer> data) {
decrementDemand();
CompletionStage<?> cs;
try {
cs = listener.onPong(webSocket, data.buffer());
} catch (Exception e) {
closeExceptionally(new RuntimeException("onPong threw an exception", e));
return;
}
follow(cs, data, "onPong");
}
@Override
public void onClose(WebSocket.CloseCode code, CharSequence reason) {
decrementDemand();
try {
close();
listener.onClose(webSocket, Optional.ofNullable(code), reason.toString());
} catch (Exception e) {
logger.log(ERROR, "onClose threw an exception", e);
}
}
}
private void follow(CompletionStage<?> cs, WSDisposable d, String source) {
if (cs == null) {
d.dispose();
} else {
cs.whenComplete((whatever, error) -> {
if (error != null) {
String m = "CompletionStage returned by " + source + " completed exceptionally";
closeExceptionally(new RuntimeException(m, error));
}
d.dispose();
});
}
}
private void decrementDemand() {
long newDemand = demand.decrementAndGet();
assert newDemand >= 0 : newDemand;
}
private RawChannel.NonBlockingEvent createChannelEvent() {
return new RawChannel.NonBlockingEvent() {
@Override
public int interestOps() {
return SelectionKey.OP_READ;
}
@Override
public void handle() {
boolean wasNotReadable = readable.compareAndSet(false, true);
assert wasNotReadable;
handler.signal();
}
@Override
public String toString() {
return "Read readiness event [" + channel + "]";
}
};
}
}