This is the second in a series of blogs on data-driven microservices design mechanisms and transaction patterns with the Oracle converged database. The first blog illustrated how to connect to an Oracle database in Java, JavaScript, Python, .NET, and Go as succinctly as possible. The goal of this second blog is to use that connection to receive and send messages with Oracle AQ (Advanced Queueing) queues and topics and conduct an update and read from the database using all of these same languages.

Advanced Queuing (AQ) is a messaging system that is part of every Oracle database edition and was first released in 2002. AQ sharded queues introduced partitioning in release 12c and is now called Transaction Event Queues (TEQ). 

Microservices increasingly use event-driven architectures for communication and related to this many data-driven systems are also employing an event sourcing pattern of one form or another. This is when data changes (eg a SQL command to "insert order") are sent via events that describe the data change (eg an "orderPlaced" event) that are received by interested services. Thus the data is sourced from the events and event sourcing in general moves the source of truth for data to the event broker.  This fits nicely with the decoupling paradigm of microservices.

It is very important to notice that there are actually two operations involved in event sourcing, the data change being made and the communication/event of that data change. There is, therefore, a transactional consideration and any inconsistency or failure causing a lack of atomicity between these two operations must be accounted for.  This is an area where TEQ has an extremely significant and unique advantage as it, the messaging/eventing system, is actually part of the database system itself and therefore can conduct both of these operations in the same local transaction and provide this atomicity guarantee. 

More details on AQ can be found at the blog Apache Kafka versus Oracle Transactional Event Queues (TEQ) as Microservices Event Mesh.

The examples provided in this blog will succinctly show the pattern just described using an order/inventory scenario:

• A local transaction is started.
• A message with a JSON payload for an order is dequeued from the orderqueue queue.
• The inventory for the item requested in the order is reduced in the inventory table and the inventory location is returned.
• A message with a JSON payload (including the inventory status and location) is enqueued on the inventoryqueue queue.
• The local transaction is committed. 

Note that the SYS.AQ$_JMS_TEXT_MESSAGE type is interoperable between all languages and supports a JSON payload making it perfect for writing microservices that work together regardless of language.

The PL/SQL to create the order and inventory queues:

  BEGIN
  DBMS_AQADM.CREATE_QUEUE_TABLE (
  queue_table          => 'ORDERQUEUETABLE',
  queue_payload_type   => 'SYS.AQ\$_JMS_TEXT_MESSAGE',
  multiple_consumers   => true,
  compatible           => '8.1');

  DBMS_AQADM.CREATE_QUEUE (
  queue_name          => 'ORDERQUEUE',
  queue_table         => 'ORDERQUEUETABLE');

  DBMS_AQADM.START_QUEUE (
  queue_name          => 'ORDERQUEUE');
  END;
  /

  BEGIN
   DBMS_AQADM.CREATE_QUEUE_TABLE (
   queue_table          => 'INVENTORYQUEUETABLE',
   queue_payload_type   => 'SYS.AQ\$_JMS_TEXT_MESSAGE',
   multiple_consumers   => true,
   compatible           => '8.1');

   DBMS_AQADM.CREATE_QUEUE (
   queue_name          => 'INVENTORYQUEUE',
   queue_table         => 'ORDERQUEUETABLE');

   DBMS_AQADM.START_QUEUE (
   queue_name          => 'INVENTORYQUEUE');
   END;
   /

The SQL to create the inventory table and insert a sample item:

   create table inventory (
     inventoryid varchar(16) PRIMARY KEY NOT NULL,
     inventorylocation varchar(32),
     inventorycount integer CONSTRAINT positive_inventory CHECK (inventorycount >= 0) );

   insert into inventory values ('sushi', '1468 WEBSTER ST,San Francisco,CA', 0);

Note that JSON can be manipulated using plain string operations, however, JSON parsers for each language are used with order and inventory object representations in the code snippets presented.  The complete source can be found at https://github.com/oracle/microservices-datadriven and you can take the "Building Microservices with converged Oracle Database Workshop" found at http://bit.ly/simplifymicroservices at any time for no charge to easily set up a full microservices environment complete with a Kubernetes cluster, 2 ATP (Autonomous Transaction Processing) Oracle databases, AQ messaging propagation, etc. in ~15 minutes!  

We'll provide just the basic facts you need... the source.

Java

    
    QueueConnectionFactory qcfact = AQjmsFactory.getQueueConnectionFactory(inventoryResource.atpInventoryPDB);
    QueueConnection qconn = qcfact.createQueueConnection(inventoryResource.inventoryuser, inventoryResource.inventorypw);
    QueueSession qsess = qconn.createQueueSession(true, Session.CLIENT_ACKNOWLEDGE);
    qconn.start();
    Queue queue = ((AQjmsSession) qsess).getQueue(inventoryResource.inventoryuser, inventoryResource.orderQueueName);
    AQjmsConsumer consumer = (AQjmsConsumer) qsess.createConsumer(queue);

    // Dequeue from order queue
    TextMessage orderMessage = (TextMessage) (consumer(-1));
    String txt = orderMessage.getText();
    Order order = JsonUtils.read(txt, Order.class);
    dbConnection = session.getDBConnection();
    OraclePreparedStatement st = (OraclePreparedStatement) dbConnection.prepareStatement("update inventory set inventorycount = inventorycount - 1 where inventoryid = ? and inventorycount > 0 returning inventorylocation into ?");
    st.setString(1, id);
    st.registerReturnParameter(2, Types.VARCHAR);

    // Update inventory and return inventory location
    int i = st.executeUpdate();
    ResultSet res = st.getReturnResultSet();
    inventorylocation = (i > 0 && res.next())?res.getString(1):"inventorydoesnotexist";
    Inventory inventory = new Inventory(orderid, itemid, inventorylocation, suggestiveSale);
    String jsonString = JsonUtils.writeValueAsString(inventory);
    Topic inventoryTopic = session.getTopic(InventoryResource.inventoryuser, InventoryResource.inventoryQueueName);
    TextMessage objmsg = session.createTextMessage();
    TopicPublisher publisher = session.createPublisher(inventoryTopic);

    // Enqueue the response on the inventory queue and commit
    publisher.publish(inventoryTopic, objmsg, DeliveryMode.PERSISTENT, 2, AQjmsConstants.EXPIRATION_NEVER);
    qsess.commit();

Python

    
    orderQueue = conn.queue(queue_owner + ".orderqueue", conn.gettype("SYS.AQ$_JMS_TEXT_MESSAGE"))
    inventoryQueue = conn.queue(queue_owner + ".inventoryqueue", conn.gettype("SYS.AQ$_JMS_TEXT_MESSAGE"))
    cursor = conn.cursor()

    # Dequeue from order queue
    conn.autocommit = False
    payload =orderQueue.deqOne().payload
    logger.debug(payload.TEXT_VC)
    orderInfo = simplejson.loads(payload.TEXT_VC)

    # Update inventory and return inventory location
    ilvar = cursor.var(str)
    cursor.execute("""update inventory set inventorycount = inventorycount - 1
         where inventoryid = :inventoryid and inventorycount > 0
         returning inventorylocation into :inventorylocation""", [orderInfo["itemid"], ilvar])

    # Enqueue the response on the inventory queue and commit
    payload = conn.gettype("SYS.AQ$_JMS_TEXT_MESSAGE").newobject()
    payload.TEXT_VC = simplejson.dumps(
        {'orderid': orderInfo["orderid"],
         'itemid': orderInfo["itemid"],
         'inventorylocation': ilvar.getvalue(0)[0] if cursor.rowcount == 1 else "inventorydoesnotexist",
         'suggestiveSale': "beer"})
    payload.TEXT_LEN = len(payload.TEXT_VC)
    inventoryQueue.enqOne(conn.msgproperties(payload = payload))
    conn.commit()

JavaScript

  
  let connection;
  let bindVariables = {};  
  let options = {};
  oracledb.autoCommit = false;
  connection = await oracledb.getConnection();
  const orderQueue = await connection.getQueue(queueConfig.orderQueue, queueOptions);

  // Dequeue from order queue
  const orderMsg = await orderQueue.deqOne();
  const orderMsgContent = JSON.parse(orderMsg.payload.TEXT_VC);
  const updateSQL = `update inventory  set inventorycount = inventorycount - 1 where 1=1 and inventoryid = :inventoryid and inventorycount > 0  returning inventorylocation into :inventorylocation`; 
  bindVariables.inventoryid = orderMsgContent.itemid
  bindVariables.inventorylocation = {
      dir: oracledb.BIND_OUT,
      type: oracledb.STRING
  };
  options.outFormat = oracledb.OUT_FORMAT_OBJECT;

  // Update inventory and return inventory location
  const queryResult = await connection.execute(updateSQL, bindVariables, options);
  if (queryResult.rowsAffected && queryResult.rowsAffected === 1) {
      location = queryResult.outBinds.inventorylocation[0];
  } else {
      location = "inventorydoesnotexist";
  }
  const inventoryQueue = await connection.getQueue(queueConfig.inventoryQueue, queueOptions);
  const inventoryMsgText = JSON.stringify({
      orderid: orderMsgContent.orderid,
      itemid: orderMsgContent.itemid,
      inventorylocation: location,
      suggestiveSale: "beer"
  });
  const inventoryMsgContent = new inventoryQueue.payloadTypeClass({
      TEXT_VC: inventoryMsgText,
      TEXT_LEN: inventoryMsgText.length
  });

  // Enqueue the response on the inventory queue and commit
  const inventoryMsg = await inventoryQueue.enqOne({payload: inventoryMsgContent, priority: 2});
  await connection.commit();

.NET

   
  
  //for .NET we will show usage of PL/SQL procedures which can be found/installed from here
  connection.Open();
  OracleTransaction tx = connection.BeginTransaction();
  OracleCommand orderReceiveMessageCommand = new OracleCommand();
  orderReceiveMessageCommand.Connection = connection;
  orderReceiveMessageCommand.CommandText = "dequeueOrderMessage";
  orderReceiveMessageCommand.CommandType = CommandType.StoredProcedure;
  OracleParameter p_orderInfoParam = new OracleParameter("p_orderInfo", OracleDbType.Varchar2, 32767);
  p_orderInfoParam.Direction = ParameterDirection.Output;
  orderReceiveMessageCommand.Parameters.Add (p_orderInfoParam);

  // Dequeue from order queue
  orderReceiveMessageCommand.ExecuteNonQuery();
  Order order = JsonConvert.DeserializeObject<Order>("" + orderReceiveMessageCommand.Parameters["p_orderInfo"].Value);
  OracleCommand checkInventoryReturnLocationCommand = new OracleCommand();
  checkInventoryReturnLocationCommand.Connection = connection;
  checkInventoryReturnLocationCommand.CommandText = "checkInventoryReturnLocation";
  checkInventoryReturnLocationCommand.CommandType =  CommandType.StoredProcedure;
  OracleParameter p_itemIdParam = new OracleParameter("p_inventoryId", OracleDbType.Varchar2, 32767);
  p_itemIdParam.Direction =ParameterDirection.Input;
  p_itemIdParam.Value = order.itemid;
  checkInventoryReturnLocationCommand.Parameters.Add (p_itemIdParam);
  OracleParameter p_inventorylocationParam = new OracleParameter("p_inventorylocation", OracleDbType.Varchar2, 32767);
  p_inventorylocationParam.Direction = ParameterDirection.Output;
  checkInventoryReturnLocationCommand.Parameters.Add (p_inventorylocationParam);

  // Update inventory and return inventory location
  checkInventoryReturnLocationCommand.ExecuteNonQuery();
  Inventory inventory = new Inventory();
  var inventoryLocation = "" + checkInventoryReturnLocationCommand.Parameters["p_inventorylocation"].Value;
  inventory.inventorylocation = inventoryLocation.Equals("null") ? "inventorydoesnotexist" : inventoryLocation;
  inventory.itemid = order.itemid;
  inventory.orderid = order.orderid;
  inventory.suggestiveSale = inventoryLocation.Equals("null") ? "" : "beer";
  string inventoryJSON = JsonConvert.SerializeObject(inventory);
  OracleCommand inventorySendMessageCommand = new OracleCommand();
  inventorySendMessageCommand.Connection = connection;
  inventorySendMessageCommand.CommandText = "enqueueInventoryMessage";
  inventorySendMessageCommand.CommandType = CommandType.StoredProcedure;
  OracleParameter p_inventoryInfoParam = new OracleParameter("p_inventoryInfo", OracleDbType.Varchar2, 32767);
  p_inventoryInfoParam.Direction = ParameterDirection.Input;
  p_inventoryInfoParam.Value = inventoryJSON;
  inventorySendMessageCommand.Parameters.Add (p_inventoryInfoParam);

  // Enqueue the response on the inventory queue and commit
  inventorySendMessageCommand.ExecuteNonQuery();
  tx.Commit();

Go

   
   tx, err := db.BeginTx(ctx, nil)
   defer tx.Rollback()
   orderqueue, err := godror.NewQueue(ctx, tx, "orderqueue", "SYS.AQ$_JMS_TEXT_MESSAGE",
      godror.WithDeqOptions(godror.DeqOptions{
         Mode:       godror.DeqRemove,
         Visibility: godror.VisibleOnCommit,
         Navigation: godror.NavNext,
         Wait:       10000,
      }))
   defer orderqueue.Close()
   msgs := make([]godror.Message, 1)

  // Dequeue from order queue
   n, err := orderqueue.Dequeue(msgs)
   textVCIntf, _ := msgs[0].Object.Get("TEXT_VC")
   textVC := fmt.Sprintf("%s", textVCIntf)
   type Order struct {
      Orderid           string
      Itemid            string
      Deliverylocation  string
      Status            string
      Inventorylocation string
      SuggestiveSale    string
   }
   order := Order{}
   json.Unmarshal([]byte(textVC), &order)
   var inventorylocation string
   sqlString := "update INVENTORY set INVENTORYCOUNT = INVENTORYCOUNT - 1 where INVENTORYID = :inventoryid and INVENTORYCOUNT > 0 returning inventorylocation into :inventorylocation"

   // Update inventory and return inventory location
   res, errFromInventoryCheck := db.Exec(sqlString, sql.Named("inventoryid", order.Itemid), sql.Named("inventorylocation", sql.Out{Dest: &inventorylocation}))
   numRows, err := res.RowsAffected()
   if inventorylocation == "" {
      inventorylocation = "inventorydoesnotexist"
   }
   type Inventory struct {
      Orderid           string
      Itemid            string
      Inventorylocation string
      SuggestiveSale    string
   }
   inventory := &Inventory{
      Orderid:           order.Orderid,
      Itemid:            order.Itemid,
      Inventorylocation: inventorylocation,
      SuggestiveSale:    "beer",
   }
   inventoryJsonData, err := json.Marshal(inventory)
   if err != nil {
      fmt.Println(err)
   }
   inventoryqueue, err := godror.NewQueue(ctx, tx, "inventoryqueue", "SYS.AQ$_JMS_TEXT_MESSAGE",
      godror.WithEnqOptions(godror.EnqOptions{
         Visibility:   godror.VisibleOnCommit, //Immediate
         DeliveryMode: godror.DeliverPersistent,
      }))
   defer inventoryqueue.Close()
   obj, err := inventoryqueue.PayloadObjectType.NewObject()
   sendmsg := godror.Message{Object: obj}
   sendmsg.Expiration = 10000
   inventoryJsonDatastr := string(inventoryJsonData)
   obj.Set("TEXT_VC",  inventoryJsonDatastr)
   obj.Set("TEXT_LEN", len(inventoryJsonDatastr))
   sendmsgs := make([]godror.Message, 1)
   sendmsgs[0] = sendmsg

   // Enqueue the response on the inventory queue and commit
   if err = inventoryqueue.Enqueue(sendmsgs); err != nil { 
      fmt.Printf("\nenqueue error:", err)
   }
   if err := tx.Commit(); err != nil {
      fmt.Printf("commit err:", err)
   }

Conclusion

We have built upon the first blog, which gave examples of how to connect to the Oracle database and containerize for microservice environments, by showing how to use Oracle AQ/TEQ for eventing as well as update and query, in the same local transaction, in order to implement a robust and simplified event-driven microservices architectural groundwork.

In the next installment of the series we will look at various frameworks of these languages and how they provide additional convenience and functionality when creating microservices using the converged Oracle Database.

Please feel free to provide any feedback here, on the workshop, on the GitHub repos, or directly. We are happy to hear from you.

I would like to thank Kuassi Mensah, Alex Keh, Christian Shay, Christopher Jones, Richard Exley, Irina Granat, and Curtis Dunkel for their development help in these languages and contributions to the workshop.