Created a decoder handler to deserialize a MemcachedResponse

Package.swift

@@ -39,6 +39,7 @@ let package = Package(
  dependencies: [
  .product(name: "NIOCore", package: "swift-nio"),
  .product(name: "NIOPosix", package: "swift-nio"),
+ .product(name: "NIOEmbedded", package: "swift-nio"),
  .product(name: "Logging", package: "swift-log"),
  ]
  ),

Sources/SwiftMemcache/MemcachedResponse.swift

@@ -0,0 +1,43 @@
+//===----------------------------------------------------------------------===//
+//
+// This source file is part of the swift-memcache-gsoc open source project
+//
+// Copyright (c) 2023 Apple Inc. and the swift-memcache-gsoc project authors
+// Licensed under Apache License v2.0
+//
+// See LICENSE.txt for license information
+// See CONTRIBUTORS.txt for the list of swift-memcache-gsoc project authors
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+//===----------------------------------------------------------------------===//
+
+struct MemcachedResponse {
+ enum ReturnCode {
+ case HD
+ case NS
+ case EX
+ case NF
+ case VA
+
+ init(_ bytes: UInt16) {
+ switch bytes {
+ case 0x4844:
+ self = .HD
+ case 0x4E53:
+ self = .NS
+ case 0x4558:
+ self = .EX
+ case 0x4E46:
+ self = .NF
+ case 0x5641:
+ self = .VA
+ default:
+ preconditionFailure("Unrecognized response code.")
+ }
+ }
+ }
+
+ var returnCode: ReturnCode
+ var dataLength: UInt64?
+}

Sources/SwiftMemcache/MemcachedResponseDecoder.swift

@@ -0,0 +1,176 @@
+//===----------------------------------------------------------------------===//
+//
+// This source file is part of the swift-memcache-gsoc open source project
+//
+// Copyright (c) 2023 Apple Inc. and the swift-memcache-gsoc project authors
+// Licensed under Apache License v2.0
+//
+// See LICENSE.txt for license information
+// See CONTRIBUTORS.txt for the list of swift-memcache-gsoc project authors
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+//===----------------------------------------------------------------------===//
+
+import NIOCore
+import NIOPosix
+
+/// Responses look like:
+///
+/// <RC> <datalen*> <flag1> <flag2> <...>\r\n
+///
+/// Where <RC> is a 2 character return code. The number of flags returned are
+/// based off of the flags supplied.
+///
+/// <datalen> is only for responses with payloads, with the return code 'VA'.
+///
+/// Flags are single character codes, ie 'q' or 'k' or 'I', which adjust the
+/// behavior of the command. If a flag requests a response flag (ie 't' for TTL
+/// remaining), it is returned in the same order as they were in the original
+/// command, though this is not strict.
+///
+/// Flags are single character codes, ie 'q' or 'k' or 'O', which adjust the
+/// behavior of a command. Flags may contain token arguments, which come after the
+/// flag and before the next space or newline, ie 'Oopaque' or 'Kuserkey'. Flags
+/// can return new data or reflect information, in the same order they were
+/// supplied in the request. Sending an 't' flag with a get for an item with 20
+/// seconds of TTL remaining, would return 't20' in the response.
+///
+/// All commands accept a tokens 'P' and 'L' which are completely ignored. The
+/// arguments to 'P' and 'L' can be used as hints or path specifications to a
+/// proxy or router inbetween a client and a memcached daemon. For example, a
+/// client may prepend a "path" in the key itself: "mg /path/foo v" or in a proxy
+/// token: "mg foo Lpath/ v" - the proxy may then optionally remove or forward the
+/// token to a memcached daemon, which will ignore them.
+///
+/// Syntax errors are handled the same as noted under 'Error strings' section
+/// below.
+///
+/// For usage examples beyond basic syntax, please see the wiki:
+/// https://github.com/memcached/memcached/wiki/MetaCommands
+struct MemcachedResponseDecoder: NIOSingleStepByteToMessageDecoder {
+ typealias InboundOut = MemcachedResponse
+
+ /// Describes the errors that can occur during the decoding process.
+ enum MemcachedDecoderError: Error {
+ /// This error is thrown when EOF is encountered but there are still
+ /// readable bytes in the buffer, which can indicate a bad message.
+ case unexpectedEOF
+
+ /// This error is thrown when EOF is encountered but there is still an expected next step
+ /// in the decoder's state machine. This error suggests that the message ended prematurely,
+ /// possibly indicating a bad message.
+ case unexpectedNextStep(NextStep)
+
+ /// This error is thrown when an unexpected character is encountered in the buffer
+ /// during the decoding process.
+ case unexpectedCharacter(UInt8)
+ }
+
+ /// The next step that the decoder will take. The value of this enum determines how the decoder
+ /// processes the current state of the ByteBuffer.
+ enum NextStep: Hashable {
+ /// The initial step.
+ case returnCode
+ /// Decode the data length, flags or check if we are the end
+ case dataLengthOrFlag(MemcachedResponse.ReturnCode)
+ /// Decode the next flag
+ case decodeNextFlag(MemcachedResponse.ReturnCode, UInt64?)
+ // TODO: Add a next step for decoding the response data if the return code is VA
+ }
+
+ /// The action that the decoder will take in response to the current state of the ByteBuffer and the `NextStep`.
+ enum NextDecodeAction {
+ /// We need more bytes to decode the next step.
+ case waitForMoreBytes
+ /// We can continue decoding.
+ case continueDecodeLoop
+ /// We have decoded the next response and need to return it.
+ case returnDecodedResponse(MemcachedResponse)
+ }
+
+ /// The next step in decoding.
+ var nextStep: NextStep = .returnCode
+
+ mutating func decode(buffer: inout ByteBuffer) throws -> InboundOut? {
+ while true {
+ switch try self.next(buffer: &buffer) {
+ case .returnDecodedResponse(let response):
+ return response
+
+ case .waitForMoreBytes:
+ return nil
+
+ case .continueDecodeLoop:
+ ()
+ }
+ }
+ }
+
+ mutating func next(buffer: inout ByteBuffer) throws -> NextDecodeAction {
+ switch self.nextStep {
+ case .returnCode:
+ guard let bytes = buffer.readInteger(as: UInt16.self) else {
+ return .waitForMoreBytes
+ }
+
+ let returnCode = MemcachedResponse.ReturnCode(bytes)
+ self.nextStep = .dataLengthOrFlag(returnCode)
+ return .continueDecodeLoop
+
+ case .dataLengthOrFlag(let returnCode):
+ if returnCode == .VA {
+ // TODO: Implement decoding of data length
+ fatalError("Decoding for VA return code is not yet implemented")
+ }
+
+ guard let nextByte = buffer.readInteger(as: UInt8.self) else {
+ return .waitForMoreBytes
+ }
+
+ if nextByte == UInt8.whitespace {
+ self.nextStep = .decodeNextFlag(returnCode, nil)
+ return .continueDecodeLoop
+ } else if nextByte == UInt8.carriageReturn {
+ guard let nextNextByte = buffer.readInteger(as: UInt8.self), nextNextByte == UInt8.newline else {
+ return .waitForMoreBytes
+ }
+ let response = MemcachedResponse(returnCode: returnCode, dataLength: nil)
+ self.nextStep = .returnCode
+ return .returnDecodedResponse(response)
+ } else {
+ throw MemcachedDecoderError.unexpectedCharacter(nextByte)
+ }
+
+ case .decodeNextFlag(let returnCode, let dataLength):
+ while let nextByte = buffer.readInteger(as: UInt8.self), nextByte != UInt8.whitespace {
+ // for now consume the byte and do nothing with it.
+ // TODO: Implement decoding of flags
+ }
+
+ let response = MemcachedResponse(returnCode: returnCode, dataLength: dataLength)
+ self.nextStep = .returnCode
+ return .returnDecodedResponse(response)
+ }
+ }
+
+ mutating func decodeLast(buffer: inout ByteBuffer, seenEOF: Bool) throws -> MemcachedResponse? {
+ // Try to decode what is left in the buffer.
+ if let output = try self.decode(buffer: &buffer) {
+ return output
+ }
+
+ guard buffer.readableBytes == 0 || seenEOF else {
+ // If there are still readable bytes left and we haven't seen an EOF
+ // then something is wrong with the message or how we called the decoder.
+ throw MemcachedDecoderError.unexpectedEOF
+ }
+
+ switch self.nextStep {
+ case .returnCode:
+ return nil
+ default:
+ throw MemcachedDecoderError.unexpectedNextStep(self.nextStep)
+ }
+ }
+}

Tests/SwiftMemcacheTests/IntegrationTest/MemcachedIntegrationTests.swift

@@ -27,7 +27,7 @@ final class MemcachedIntegrationTest: XCTestCase {
  self.channel = ClientBootstrap(group: self.group)
  .channelOption(ChannelOptions.socket(SocketOptionLevel(SOL_SOCKET), SO_REUSEADDR), value: 1)
  .channelInitializer { channel in
- channel.pipeline.addHandler(MessageToByteHandler(MemcachedRequestEncoder()))
+ return channel.pipeline.addHandlers([MessageToByteHandler(MemcachedRequestEncoder()), ByteToMessageHandler(MemcachedResponseDecoder())])
  }
  }
 
@@ -36,6 +36,21 @@ final class MemcachedIntegrationTest: XCTestCase {
  super.tearDown()
  }
 
+ class ResponseHandler: ChannelInboundHandler {
+ typealias InboundIn = MemcachedResponse
+
+ let p: EventLoopPromise<MemcachedResponse>
+
+ init(p: EventLoopPromise<MemcachedResponse>) {
+ self.p = p
+ }
+
+ func channelRead(context: ChannelHandlerContext, data: NIOAny) {
+ let response = self.unwrapInboundIn(data)
+ self.p.succeed(response)
+ }
+ }
+
  func testConnectionToMemcachedServer() throws {
  do {
  let connection = try channel.connect(host: "memcached", port: 11211).wait()
@@ -47,15 +62,24 @@ final class MemcachedIntegrationTest: XCTestCase {
  let command = MemcachedRequest.SetCommand(key: "foo", value: buffer)
  let request = MemcachedRequest.set(command)
 
- // Write the request to the connection and wait for the result
- connection.writeAndFlush(request).whenComplete { result in
- switch result {
- case .success:
- print("Request successfully sent to the server.")
- case .failure(let error):
- XCTFail("Failed to send request: \(error)")
- }
- }
+ // Write the request to the connection
+ _ = connection.write(request)
+
+ // Prepare the promise for the response
+ let promise = connection.eventLoop.makePromise(of: MemcachedResponse.self)
+ let responseHandler = ResponseHandler(p: promise)
+ _ = connection.pipeline.addHandler(responseHandler)
+
+ // Flush and then read the response from the server
+ connection.flush()
+ connection.read()
+
+ // Wait for the promise to be fulfilled
+ let response = try promise.futureResult.wait()
+
+ // Check the response from the server.
+ print("Response return code: \(response.returnCode)")
+
  } catch {
  XCTFail("Failed to connect to Memcached server: \(error)")
  }

Tests/SwiftMemcacheTests/UnitTest/MemcachedResponseDecoderTests.swift

@@ -0,0 +1,97 @@
+//===----------------------------------------------------------------------===//
+//
+// This source file is part of the swift-memcache-gsoc open source project
+//
+// Copyright (c) 2023 Apple Inc. and the swift-memcache-gsoc project authors
+// Licensed under Apache License v2.0
+//
+// See LICENSE.txt for license information
+// See CONTRIBUTORS.txt for the list of swift-memcache-gsoc project authors
+//
+// SPDX-License-Identifier: Apache-2.0
+//
+//===----------------------------------------------------------------------===//
+
+import NIOCore
+import NIOEmbedded
+@testable import SwiftMemcache
+import XCTest
+
+final class MemcachedResponseDecoderTests: XCTestCase {
+ var decoder: MemcachedResponseDecoder!
+
+ override func setUp() {
+ super.setUp()
+ self.decoder = MemcachedResponseDecoder()
+ }
+
+ func makeMemcachedResponseByteBuffer(from response: MemcachedResponse) -> ByteBuffer {
+ var buffer = ByteBufferAllocator().buffer(capacity: 8)
+ var returnCode: UInt16 = 0
+
+ // Convert the return code enum to UInt16 then write it to the buffer.
+ switch response.returnCode {
+ case .HD:
+ returnCode = 0x4844
+ case .NS:
+ returnCode = 0x4E53
+ case .EX:
+ returnCode = 0x4558
+ case .NF:
+ returnCode = 0x4E46
+ case .VA:
+ returnCode = 0x5641
+ }
+
+ buffer.writeInteger(returnCode)
+
+ // If there's a data length, write it to the buffer.
+ if let dataLength = response.dataLength, response.returnCode == .VA {
+ buffer.writeInteger(UInt8.whitespace, as: UInt8.self)
+ buffer.writeInteger(dataLength, as: UInt64.self)
+ }
+
+ buffer.writeBytes([UInt8.carriageReturn, UInt8.newline])
+ return buffer
+ }
+
+ func testDecodeResponse(buffer: inout ByteBuffer, expectedReturnCode: MemcachedResponse.ReturnCode) throws {
+ // Pass our response through the decoder
+ var output: MemcachedResponse? = nil
+ do {
+ output = try self.decoder.decode(buffer: &buffer)
+ } catch {
+ XCTFail("Decoding failed with error: \(error)")
+ }
+ // Check the decoded response
+ if let decoded = output {
+ XCTAssertEqual(decoded.returnCode, expectedReturnCode)
+ } else {
+ XCTFail("Failed to decode the inbound response.")
+ }
+ }
+
+ func testDecodeStoredResponse() throws {
+ let storedResponse = MemcachedResponse(returnCode: .HD, dataLength: nil)
+ var buffer = self.makeMemcachedResponseByteBuffer(from: storedResponse)
+ try self.testDecodeResponse(buffer: &buffer, expectedReturnCode: .HD)
+ }
+
+ func testDecodeNotStoredResponse() throws {
+ let notStoredResponse = MemcachedResponse(returnCode: .NS, dataLength: nil)
+ var buffer = self.makeMemcachedResponseByteBuffer(from: notStoredResponse)
+ try self.testDecodeResponse(buffer: &buffer, expectedReturnCode: .NS)
+ }
+
+ func testDecodeExistResponse() throws {
+ let existResponse = MemcachedResponse(returnCode: .EX, dataLength: nil)
+ var buffer = self.makeMemcachedResponseByteBuffer(from: existResponse)
+ try self.testDecodeResponse(buffer: &buffer, expectedReturnCode: .EX)
+ }
+
+ func testDecodeNotFoundResponse() throws {
+ let notFoundResponse = MemcachedResponse(returnCode: .NF, dataLength: nil)
+ var buffer = self.makeMemcachedResponseByteBuffer(from: notFoundResponse)
+ try self.testDecodeResponse(buffer: &buffer, expectedReturnCode: .NF)
+ }
+}