raboof/pekko
1
0
Fork 0
Code Issues Pull requests Projects Releases 6 Packages Wiki Activity Actions 10

A few ByteString.size switched to ByteString.length #30342

Browse source
This commit is contained in:
Johan Andrén 2021-07-02 15:53:52 +02:00
parent 9657f8430a
commit 0dfa9e7675
5 changed files with 15 additions and 15 deletions
Download patch file Download diff file Expand all files Collapse all files

6
akka-remote/src/main/scala/akka/remote/Endpoint.scala
View file

@ -496,7 +496,7 @@ private[remote] class ReliableDeliverySupervisor(
// If we have not confirmed the remote UID we cannot transfer the system message at this point just buffer it.
// GotUid will kick resendAll() causing the messages to be properly written.
// Flow control by not sending more when we already have many outstanding.
if (uidConfirmed && resendBuffer.nonAcked.size <= settings.SysResendLimit)
if (uidConfirmed && resendBuffer.nonAcked.length <= settings.SysResendLimit)
writer ! sequencedSend
} else writer ! send
@ -1130,7 +1130,7 @@ private[remote] class EndpointReader(
override def receive: Receive = {
case Disassociated(info) => handleDisassociated(info)
case InboundPayload(p) if p.size <= transport.maximumPayloadBytes =>
case InboundPayload(p) if p.length <= transport.maximumPayloadBytes =>
val (ackOption, msgOption) = tryDecodeMessageAndAck(p)
for (ack <- ackOption; reliableDelivery <- reliableDeliverySupervisor) reliableDelivery ! ack
@ -1180,7 +1180,7 @@ private[remote] class EndpointReader(
case StopReading(writer, replyTo) =>
replyTo ! StoppedReading(writer)
case InboundPayload(p) if p.size <= transport.maximumPayloadBytes =>
case InboundPayload(p) if p.length <= transport.maximumPayloadBytes =>
val (ackOption, msgOption) = tryDecodeMessageAndAck(p)
for (ack <- ackOption; reliableDelivery <- reliableDeliverySupervisor) reliableDelivery ! ack

2
akka-remote/src/main/scala/akka/remote/artery/tcp/TcpFraming.scala
View file

@ -69,7 +69,7 @@ import akka.util.ByteString
case object ReadMagic extends Step {
override def parse(reader: ByteReader): ParseResult[EnvelopeBuffer] = {
val magic = reader.take(TcpFraming.Magic.size)
val magic = reader.take(TcpFraming.Magic.length)
if (magic == TcpFraming.Magic)
ParseResult(None, ReadStreamId)
else

4
akka-stream/src/main/scala/akka/stream/impl/io/ByteStringParser.scala
View file

@ -185,8 +185,8 @@ import akka.util.ByteString
private[this] var off = 0
def hasRemaining: Boolean = off < input.size
def remainingSize: Int = input.size - off
def hasRemaining: Boolean = off < input.length
def remainingSize: Int = input.length - off
def currentOffset: Int = off

4
akka-stream/src/main/scala/akka/stream/impl/io/compression/DeflateCompressor.scala
View file

@ -19,12 +19,12 @@ import akka.util.{ ByteString, ByteStringBuilder }
protected lazy val deflater = new Deflater(level, nowrap)
override final def compressAndFlush(input: ByteString): ByteString = {
val buffer = newTempBuffer(input.size)
val buffer = newTempBuffer(input.length)
compressWithBuffer(input, buffer) ++ flushWithBuffer(buffer)
}
override final def compressAndFinish(input: ByteString): ByteString = {
val buffer = newTempBuffer(input.size)
val buffer = newTempBuffer(input.length)
compressWithBuffer(input, buffer) ++ finishWithBuffer(buffer)
}

14
akka-stream/src/main/scala/akka/stream/scaladsl/Framing.scala
View file

@ -265,27 +265,27 @@ object Framing {
s"Read ${possibleMatchPos - previous} bytes " +
s"which is more than $maximumLineBytes without seeing a line terminator"))
} else if (possibleMatchPos == -1) {
if (buffer.size - previous > maximumLineBytes)
if (buffer.length - previous > maximumLineBytes)
failStage(
new FramingException(
s"Read ${buffer.size - previous} bytes " +
s"Read ${buffer.length - previous} bytes " +
s"which is more than $maximumLineBytes without seeing a line terminator"))
else {
// No matching character, we need to accumulate more bytes into the buffer
nextPossibleMatch = buffer.size
nextPossibleMatch = buffer.length
doParse()
}
} else if (possibleMatchPos + separatorBytes.size > buffer.size) {
} else if (possibleMatchPos + separatorBytes.length > buffer.length) {
// We have found a possible match (we found the first character of the terminator
// sequence) but we don't have yet enough bytes. We remember the position to
// retry from next time.
nextPossibleMatch = possibleMatchPos
doParse()
} else if (buffer.slice(possibleMatchPos, possibleMatchPos + separatorBytes.size) == separatorBytes) {
} else if (buffer.slice(possibleMatchPos, possibleMatchPos + separatorBytes.length) == separatorBytes) {
// Found a match, mark start and end position and iterate if possible
indices += (previous -> possibleMatchPos)
nextPossibleMatch = possibleMatchPos + separatorBytes.size
if (nextPossibleMatch == buffer.size || indices.isFull) {
nextPossibleMatch = possibleMatchPos + separatorBytes.length
if (nextPossibleMatch == buffer.length || indices.isFull) {
doParse()
} else {
searchIndices()