From c9a4f064ca1fe8fb06b67f7ac09ba2777bbb5634 Mon Sep 17 00:00:00 2001 From: Oleh Dokuka Date: Mon, 4 May 2020 15:32:20 +0300 Subject: [PATCH] relaxed stream id supplement (#814) --- .../java/io/rsocket/core/StreamIdSupplier.java | 17 +++++++++++------ 1 file changed, 11 insertions(+), 6 deletions(-) diff --git a/rsocket-core/src/main/java/io/rsocket/core/StreamIdSupplier.java b/rsocket-core/src/main/java/io/rsocket/core/StreamIdSupplier.java index 70734b8c0..7f4d7b7b3 100644 --- a/rsocket-core/src/main/java/io/rsocket/core/StreamIdSupplier.java +++ b/rsocket-core/src/main/java/io/rsocket/core/StreamIdSupplier.java @@ -1,5 +1,5 @@ /* - * Copyright 2015-2018 the original author or authors. + * Copyright 2015-2020 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -16,14 +16,11 @@ package io.rsocket.core; import io.netty.util.collection.IntObjectMap; -import java.util.concurrent.atomic.AtomicLongFieldUpdater; final class StreamIdSupplier { private static final int MASK = 0x7FFFFFFF; - private static final AtomicLongFieldUpdater STREAM_ID = - AtomicLongFieldUpdater.newUpdater(StreamIdSupplier.class, "streamId"); - private volatile long streamId; + private long streamId; // Visible for testing StreamIdSupplier(int streamId) { @@ -38,10 +35,18 @@ static StreamIdSupplier serverSupplier() { return new StreamIdSupplier(0); } + /** + * This methods provides new stream id and ensures there is no intersections with already running + * streams. This methods is not thread-safe. + * + * @param streamIds currently running streams store + * @return next stream id + */ int nextStreamId(IntObjectMap streamIds) { int streamId; do { - streamId = (int) STREAM_ID.addAndGet(this, 2) & MASK; + this.streamId += 2; + streamId = (int) (this.streamId & MASK); } while (streamId == 0 || streamIds.containsKey(streamId)); return streamId; }