Package org.acemq.amqp.patterns

Class JdbcSchemaRegistry

java.lang.Object
org.acemq.amqp.patterns.JdbcSchemaRegistry
All Implemented Interfaces:
SchemaRegistry
public final class JdbcSchemaRegistry extends Object implements SchemaRegistry
A schema registry in a database table.

The in-memory one is fine for a test and useless for anything else: identifiers must be stable forever, because a message published today may be read next year by a consumer looking up the schema it was written with. A registry that hands out fresh identifiers on restart makes every message written before the restart unreadable, and does it silently.

This is the smallest thing that fixes that. It is not Confluent's registry and does not try to be: there is no compatibility checking, no versioning UI, no REST API. What it does is remember which integer stands for which schema, across restarts and across instances, which is the part the wire format depends on. 


 JdbcSchemaRegistry registry = new JdbcSchemaRegistry(dataSource);
 registry.createSchemaIfAbsent();

 mq.publisher("orders", "order.placed", GenericRecord.class)
         .as(AvroCodec.registered(registry))
         .send(order);

Both directions are cached in memory and never invalidated, because neither answer can change: an identifier stands for one schema forever, and a schema keeps the identifier it was given. Without the cache this would be a database round trip per message, which is not a registry anyone would keep.

  • Constructor Details

    • JdbcSchemaRegistry

      public JdbcSchemaRegistry(DataSource dataSource)
      Parameters:
      dataSource - where the table lives

    • JdbcSchemaRegistry

      public JdbcSchemaRegistry(DataSource dataSource, String table)
      Parameters:
      dataSource - where the table lives
      table - table name; must be a plain SQL identifier

  • Method Details

    • createSchemaIfAbsent

      public void createSchemaIfAbsent()
      Creates the table if it is not already there.

      For development and tests. In production the table belongs in the migration tool that owns the rest of the schema.

      Throws:
      AceMqException - if the schema cannot be created

    • idFor

      public int idFor(SchemaDefinition schema)
      Specified by:
      idFor in interface SchemaRegistry
      Parameters:
      schema - the schema a message is about to be written with
      Returns:
      its identifier, registering it if the registry has not seen it. Registering the same schema twice must return the same identifier

    • schemaFor

      public SchemaDefinition schemaFor(int id)
      Specified by:
      schemaFor in interface SchemaRegistry
      Parameters:
      id - an identifier read from a message
      Returns:
      the schema it stands for

    • size

      public int size()
      Returns:
      how many schemas this registry holds

    • toString

      public String toString()
      Overrides:
      toString in class Object