go-ora: Regression since v2.7.23 when using DBMS_AQ.DEQUEUE_ARRAY

dequeue_array.go

package main

import (
	"database/sql"
	"flag"
	"fmt"
	"os"
	"time"

	go_ora "github.com/sijms/go-ora/v2"
)

type test1 struct {
	Name string `udt:"name"`
}

func createTypes(conn *sql.DB) error {
	sqlText := `CREATE TYPE test1 AS OBJECT ( name varchar2(256) )`
	_, err := conn.Exec(sqlText)
	if err != nil {
		return err
	}

	sqlText = `CREATE OR REPLACE TYPE test1collection AS TABLE OF test1`
	_, err = conn.Exec(sqlText)
	if err != nil {
		return err
	}

	return nil
}

func dropTypes(conn *sql.DB) error {
	sqlText := `DROP TYPE test1collection`
	_, err := conn.Exec(sqlText)
	if err != nil {
		return err
	}

	sqlText = `DROP TYPE test1`
	_, err = conn.Exec(sqlText)
	if err != nil {
		return err
	}

	return nil
}

func setupQueue(conn *sql.DB) error {
	sqlText := `BEGIN DBMS_AQADM.CREATE_QUEUE_TABLE ( queue_table => 'test1table', queue_payload_type => 'test1' ); END;`
	_, err := conn.Exec(sqlText)
	if err != nil {
		return err
	}

	sqlText = `BEGIN DBMS_AQADM.CREATE_QUEUE ( queue_name => 'test1queue', queue_table => 'test1table' ); END;`
	_, err = conn.Exec(sqlText)
	if err != nil {
		return err
	}

	sqlText = `BEGIN DBMS_AQADM.START_QUEUE ( queue_name => 'test1queue', enqueue => TRUE ); END;`
	_, err = conn.Exec(sqlText)
	if err != nil {
		return err
	}

	return nil
}

func stopQueue(conn *sql.DB) error {
	sqlText := `BEGIN DBMS_AQADM.STOP_QUEUE(queue_name => 'test1queue'); END;`
	_, err := conn.Exec(sqlText)
	if err != nil {
		return err
	}

	sqlText = `BEGIN DBMS_AQADM.DROP_QUEUE(queue_name => 'test1queue'); END;`
	_, err = conn.Exec(sqlText)
	if err != nil {
		return err
	}

	sqlText = `BEGIN DBMS_AQADM.DROP_QUEUE_TABLE(queue_table => 'test1table'); END;`
	_, err = conn.Exec(sqlText)
	if err != nil {
		return err
	}

	return nil
}

func enqueue(conn *sql.DB, message test1) error {
	sqlText := `
DECLARE
	enqueueOptions			DBMS_AQ.enqueue_options_t;
	messageProperties		DBMS_AQ.message_properties_t;
	msgID_raw				RAW(100);
BEGIN
	DBMS_AQ.ENQUEUE(
		queue_name => 'test1queue',
		enqueue_options => enqueueOptions,
		message_properties => messageProperties,
		payload => :1,
		msgid => msgID_raw
	);
END;`
	_, err := conn.Exec(sqlText, message)
	if err != nil {
		return err
	}

	return nil
}

func dequeueArray(conn *sql.DB, arraySize int, waitTime time.Duration) ([]test1, error) {
	sqlText := fmt.Sprintf(`
DECLARE
	dequeueOptions			DBMS_AQ.dequeue_options_t;
	messagePropertiesArray	DBMS_AQ.message_properties_array_t;
	msgIDArray				DBMS_AQ.msgid_array_t;
BEGIN
	dequeueOptions.WAIT := %d;
	:1 := DBMS_AQ.DEQUEUE_ARRAY(
		queue_name => 'test1queue',
		dequeue_options => dequeueOptions,
		array_size => %d,
		message_properties_array => messagePropertiesArray,
		payload_array => :2,
		msgid_array => msgIDArray
	);
END;`, (waitTime / time.Second), arraySize)
	var nMessages sql.NullInt64
	var messages []test1
	_, err := conn.Exec(sqlText, sql.Out{Dest: &nMessages}, sql.Out{Dest: &messages})
	if err != nil {
		return nil, err
	}

	return messages, nil
}

func usage() {
	fmt.Println()
	fmt.Println("dequeue_array")
	fmt.Println()
	fmt.Println("Usage:")
	fmt.Println(`  dequeue_array -server server_url`)
	flag.PrintDefaults()
	fmt.Println()
	fmt.Println("Example:")
	fmt.Println(`  dequeue_array -server "oracle://user:pass@server/service_name"`)
	fmt.Println()
}

func main() {
	var (
		server string
	)

	flag.StringVar(&server, "server", "", "Server's URL, oracle://user:pass@server/service_name")
	flag.Parse()

	connStr := os.ExpandEnv(server)
	if connStr == "" {
		fmt.Println("Missing -server option")
		usage()
		os.Exit(1)
	}
	fmt.Println("Connected")
	conn, err := sql.Open("oracle", connStr)
	if err != nil {
		fmt.Println("Can't create connection: ", err)
		return
	}
	defer func() {
		err = conn.Close()
		if err != nil {
			fmt.Println("Can't close connection: ", err)
		}
	}()

	err = conn.Ping()
	if err != nil {
		fmt.Println("Can't ping connection: ", err)
		return
	}

	err = createTypes(conn)
	if err != nil {
		fmt.Println("Can't create types: ", err)
		return
	}
	defer func() {
		err := dropTypes(conn)
		if err != nil {
			fmt.Println("Can't drop types: ", err)
		}
	}()

	if err := go_ora.RegisterType(conn, "test1", "test1collection", test1{}); err != nil {
		fmt.Println("Failed to register type", err)
		return
	}

	err = setupQueue(conn)
	if err != nil {
		fmt.Println("Can't setup queue: ", err)
		return
	}
	defer func() {
		err := stopQueue(conn)
		if err != nil {
			fmt.Println("Can't stop queue: ", err)
		}
	}()

	err = enqueue(conn, test1{Name: "Test 1"})
	if err != nil {
		fmt.Println("Can't enqueue message: ", err)
		return
	}

	messages, err := dequeueArray(conn, 2, 5*time.Second)
	if err != nil {
		fmt.Println("Failed to dequeue array", err)
		return
	}

	fmt.Printf("Dequeued %d messages:\n", len(messages))
	for _, msg := range messages {
		fmt.Printf("  Message: %+v\n", msg)
	}
}

go.mod

module gist.github.com/robstradling/7046c09dbd173756ce9c6e642c2b2050

go 1.20

require github.com/sijms/go-ora/v2 v2.7.23

output_v2.7.22.txt

Connected
Dequeued 1 messages:
  Message: {Name:Test 1}

output_v2.7.23.txt

Connected
Failed to dequeue array ORA-06502: PL/SQL: numeric or value error
ORA-06512: at line 8