DEV Community: Pavel

Spring Boot 3 with MongoDB transactions

Pavel — Thu, 07 Sep 2023 20:08:45 +0000

Transactions won't work without next things:

replica set

Since transactions are built on concepts of logical sessions they require mechanics (like oplog) which are only available in replica set environment.

You can always convert a standalone to a single noded replica set and transactions will work with this one node.

https://docs.mongodb.com/manual/tutorial/convert-standalone-to-replica-set/

Reference

MongoTransactionManager

Unless you specify a MongoTransactionManager within your application context, transaction support is DISABLED. You can use setSessionSynchronization(ALWAYS) to participate in ongoing non-native MongoDB transactions.

Reference

That's main reason why newly created Spring Boot project with MongoDB will fail with next error:

java.lang.IllegalStateException: Failed to retrieve PlatformTransactionManager for @Transactional test:

What to do:

Create replica set
Configure MongoTransactionManager as PlatformTransactionManager
Update application properties
Fix DataMongoTest tests

Create replica set

Choose preferred way to create replica set:

Configure PlatformTransactionManager

import org.jetbrains.annotations.NotNull;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.mongodb.MongoDatabaseFactory;
import org.springframework.data.mongodb.MongoTransactionManager;
import org.springframework.data.mongodb.config.AbstractMongoClientConfiguration;

@Configuration
public class MongoClientConfiguration extends AbstractMongoClientConfiguration {

    @Value("${spring.data.mongodb.database}")
    private String databaseName;

    @Bean
    public MongoTransactionManager transactionManager(MongoDatabaseFactory mongoDatabaseFactory) {
        return new MongoTransactionManager(mongoDatabaseFactory);
    }

    @NotNull
    @Override
    protected String getDatabaseName() {
        return databaseName;
    }

}

Update application properties

spring.data.mongodb.replica-set-name=myReplicaSet

Fix DataMongoTest tests

Looks like @DataMongoTest still not activates autoconfiguration for MongoDB transactions.

References:

Proposal to import TransactionAutoConfiguration didn't help me

@ImportAutoConfiguration(TransactionAutoConfiguration.class)

So I decided to replace it with MongoDBConfig import

import app.config.MongoDBConfig;
import org.junit.jupiter.api.Test;
import org.springframework.boot.test.autoconfigure.data.mongo.DataMongoTest;
import org.springframework.context.annotation.Import;
import org.springframework.transaction.annotation.Transactional;

@DataMongoTest
@Transactional
@Import(MongoDBConfig.class)
public class MongoTest {

    @Test
    public void testTransaction() {

    }

}

References

How to document SSE app

Pavel — Thu, 07 Sep 2023 00:18:49 +0000

Most interesting moment for me is to compare OpenAPI specification with AsyncAPI specification in case of:

re-using of common parts
describing of SSE application

and compare results

What will we document

Service which will broadcast received messages through an SSE connection to any subscribed user

Common part

Let's collect all required schemas in one file:

schemas.json

{
  "schemas": {
    "Message": {
      "type": "object",
      "additionalProperties": false,
      "required": [
        "message",
        "receivedAt"
      ],
      "properties": {
        "message": {
          "type": "string",
          "example": "broadcast this message :rocket:"
        },
        "receivedAt": {
          "type": "string",
          "format": "date-time",
          "example": "2023-08-31T15:28:21.283+00:00",
          "description": "Date-time when application received this message"
        }
      }
    },
    "MessageToBroadcast": {
      "type": "object",
      "additionalProperties": false,
      "required": [
        "message"
      ],
      "properties": {
        "message": {
          "type": "string",
          "example": "broadcast this message :rocket:",
          "description": "Ordinary text which will be send"
        }
      }
    }
  }
}

OpenAPI

Broadcast messages

Just basic declaration, nothing special at all

{
  "openapi": "3.0.3",
  "info": {
    "version": "1.0.0",
    "title": "Messages API",
    "description": "Broadcasts received messages through an SSE connection to any subscribed user"
  },
  "servers": [
    {
      "url": "http://localhost:8080"
    }
  ],
  "paths": {
    "/messages": {
      "post": {
        "summary": "Broadcast message",
        "operationId": "broadcastMessage",
        "description": "Send message to broadcast to any subscribed user",
        "tags": [
          "messages"
        ],
        "requestBody": {
          "description": "Message to broadcast",
          "required": true,
          "content": {
            "application/json": {
              "schema": {
                "$ref": "../schemas.json#/schemas/MessageToBroadcast"
              },
              "example": "broadcast this message :rocket:"
            }
          }
        },
        "responses": {
          "200": {
            "description": "message received"
          }
        }
      }
    }
  }
}

Subscribe to messages stream

More complicated part. Unfortunately, community still figuring out how to describe events by OpenAPI in the right manner.

We can use this GitHub issue as a reference

{
  "openapi": "3.0.3",
  "info": {
    "version": "1.0.0",
    "title": "Messages stream API",
    "description": "Broadcasts received messages through an SSE connection to any subscribed user"
  },
  "servers": [
    {
      "url": "http://localhost:8080"
    }
  ],
  "paths": {
    "/messages": {
      "get": {
        "summary": "Subscribe to stream of messages",
        "operationId": "messagesStreamSubscribe",
        "description": "Receive all incoming messages",
        "tags": [
          "messages"
        ],
        "responses": {
          "200": {
            "description": "Stream of messages",
            "headers": {
              "X-SSE-Content-Type": {
                "schema": {
                  "type": "string",
                  "enum": ["application/json"]
                }
              },
              "transfer-encoding": {
                "schema": {
                  "type": "string",
                  "enum": ["chunked"]
                }
              }
            },
            "content": {
              "text/event-stream": {
                "schema": {
                  "$ref": "#/components/schemas/MessagesStream"
                }
              }
            }
          }
        }
      }
    }
  },
  "components": {
    "schemas": {
      "MessagesStream": {
        "type": "array",
        "format": "event-stream",
        "items": {
          "$ref": "../schemas.json#/schemas/Message"
        }
      }
    }
  }
}

Pub + Sub

{
  "openapi": "3.0.3",
  "info": {
    "version": "1.0.0",
    "title": "Messages API",
    "description": "Broadcasts received messages through an SSE connection to any subscribed user"
  },
  "servers": [
    {
      "url": "http://localhost:8080"
    }
  ],
  "paths": {
    "/messages": {
      "get": {
        "summary": "Subscribe to stream of messages",
        "operationId": "messagesStreamSubscribe",
        "description": "Receive all incoming messages",
        "tags": [
          "messages"
        ],
        "responses": {
          "200": {
            "description": "Stream of messages",
            "headers": {
              "X-SSE-Content-Type": {
                "schema": {
                  "type": "string",
                  "enum": ["application/json"]
                }
              },
              "transfer-encoding": {
                "schema": {
                  "type": "string",
                  "enum": ["chunked"]
                }
              }
            },
            "content": {
              "text/event-stream": {
                "schema": {
                  "$ref": "#/components/schemas/MessagesStream"
                }
              }
            }
          }
        }
      },
      "post": {
        "summary": "Broadcast message",
        "operationId": "broadcastMessage",
        "description": "Send message to broadcast to any subscribed user",
        "tags": [
          "messages"
        ],
        "requestBody": {
          "description": "Message to broadcast",
          "required": true,
          "content": {
            "application/json": {
              "schema": {
                "$ref": "../schemas.json#/schemas/MessageToBroadcast"
              },
              "example": "broadcast this message :rocket:"
            }
          }
        },
        "responses": {
          "200": {
            "description": "message received"
          }
        }
      }
    }
  },
  "components": {
    "schemas": {
      "MessagesStream": {
        "type": "array",
        "format": "event-stream",
        "items": {
          "$ref": "../schemas.json#/schemas/Message"
        }
      }
    }
  }
}

AsyncAPI

Broadcast messages

{
  "asyncapi": "2.6.0",
  "info": {
    "title": "Messages stream API",
    "description": "Broadcasts received messages through an SSE connection to any subscribed user",
    "version": "1.0.0"
  },
  "servers": {
    "dev": {
      "url": "http://localhost:8080",
      "protocol": "http"
    }
  },
  "channels": {
    "/messages": {
      "description": "Broadcast message",
      "publish": {
        "description": "Send message to broadcast to any subscribed user",
        "message": {
          "bindings": {
            "http": {
              "headers": {
                "type": "object",
                "additionalProperties": false,
                "required": [
                  "Content-Type"
                ],
                "properties": {
                  "Content-Type": {
                    "type": "string",
                    "enum": ["application/json"]
                  }
                }
              }
            }
          },
          "$ref": "#/components/messages/MessageToBroadcast"
        },
        "bindings": {
          "http": {
            "type": "request",
            "method": "POST"
          }
        }
      }
    }
  },
  "components": {
    "messages": {
      "MessageToBroadcast": {
        "payload": {
          "$ref": "../schemas.json#/schemas/MessageToBroadcast"
        }
      }
    }
  }
}

Subscribe to messages stream

{
  "asyncapi": "2.6.0",
  "info": {
    "title": "Messages stream API",
    "description": "Broadcasts received messages through an SSE connection to any subscribed user",
    "version": "1.0.0"
  },
  "servers": {
    "dev": {
      "url": "http://localhost:8080",
      "protocol": "http"
    }
  },
  "channels": {
    "/messages": {
      "description": "Subscribe to stream of messages",
      "subscribe": {
        "description": "Receive all incoming messages",
        "message": {
          "bindings": {
            "http": {
              "headers": {
                "type": "object",
                "additionalProperties": false,
                "required": [
                  "Content-Type", "X-SSE-Content-Type", "transfer-encoding"
                ],
                "properties": {
                  "Content-Type": {
                    "type": "string",
                    "enum": ["text/event-stream"]
                  },
                  "X-SSE-Content-Type": {
                    "type": "string",
                    "enum": ["application/json"]
                  },
                  "transfer-encoding": {
                    "type": "string",
                    "enum": ["chunked"]
                  }
                }
              }
            }
          },
          "$ref": "#/components/messages/Message"
        },
        "bindings": {
          "http": {
            "type": "request",
            "method": "GET"
          }
        }
      }
    }
  },
  "components": {
    "messages": {
      "Message": {
        "payload": {
          "$ref": "../schemas.json#/schemas/Message"
        }
      }
    }
  }
}

Pub + Sub

{
  "asyncapi": "2.6.0",
  "info": {
    "title": "Messages stream API",
    "description": "Broadcasts received messages through an SSE connection to any subscribed user",
    "version": "1.0.0"
  },
  "servers": {
    "dev": {
      "url": "http://localhost:8080",
      "protocol": "http"
    }
  },
  "channels": {
    "/messages": {
      "description": "Broadcast message",
      "publish": {
        "description": "Send message to broadcast to any subscribed user",
        "message": {
          "bindings": {
            "http": {
              "headers": {
                "type": "object",
                "additionalProperties": false,
                "required": [
                  "Content-Type"
                ],
                "properties": {
                  "Content-Type": {
                    "type": "string",
                    "enum": ["application/json"]
                  }
                }
              }
            }
          },
          "$ref": "#/components/messages/MessageToBroadcast"
        },
        "bindings": {
          "http": {
            "type": "request",
            "method": "POST"
          }
        }
      },
      "subscribe": {
        "description": "Receive all incoming messages",
        "message": {
          "bindings": {
            "http": {
              "headers": {
                "type": "object",
                "additionalProperties": false,
                "required": [
                  "Content-Type", "X-SSE-Content-Type", "transfer-encoding"
                ],
                "properties": {
                  "Content-Type": {
                    "type": "string",
                    "enum": ["text/event-stream"]
                  },
                  "X-SSE-Content-Type": {
                    "type": "string",
                    "enum": ["application/json"]
                  },
                  "transfer-encoding": {
                    "type": "string",
                    "enum": ["chunked"]
                  }
                }
              }
            }
          },
          "$ref": "#/components/messages/Message"
        },
        "bindings": {
          "http": {
            "type": "request",
            "method": "GET"
          }
        }
      }
    }
  },
  "components": {
    "messages": {
      "MessageToBroadcast": {
        "payload": {
          "$ref": "../schemas.json#/schemas/MessageToBroadcast"
        }
      },
      "Message": {
        "payload": {
          "$ref": "../schemas.json#/schemas/Message"
        }
      }
    }
  }
}

Resume

Both specifications allow to you to describe Pub and Sub for SSE app, but with some tradeoffs

OpenAPI can't offer canonical way how to describe stream.

For example:

{
  "type": "array",
  "format": "event-stream",
  "items": {
    "$ref": "../schemas.json#/schemas/Message"
  }
}

What does it mean? Stream of messages? Stream of arrays with messages?

From other side AsyncAPI as expected can't offer flexible syntax for description of HTTP requests - headers, params location, status codes

It's up to you to choose which specification to use, but looks like it's better to use them both, instead of reinvent the wheel: