Callback

CHANGELOG.md

@@ -1,5 +1,12 @@
 # Changelog
 
+## December 2018 - v3.3.0
+* All relevant API calls now automatically set FAIL_IF_QUIESCING
+* Samples updated to use "defer" instead of just suggesting it
+* Add support for MQCB/MQCTL callback functions
+* Add support for MQBEGIN transaction management
+* Add Dead Letter Header parser
+
 ## November 2018 - v3.2.0
 * Added GetPlatform to mqmetric so it can be used as a label/tag in collectors
 * Added sample programs demonstrating specific operations such as put/get of message

README.md

@@ -123,12 +123,7 @@ At this point, you should have a compiled copy of the program in `$GOPATH/bin`.
 
 ## Limitations
 
-Almost all of the MQI verbs are now available through the `ibmmq` package.
-Currently unavailable verbs include:
-
-* MQCB/MQCTL
-
-There are also no structure handlers for message headers such as MQRFH2 or MQDLH.
+All regular MQI verbs are now available through the `ibmmq` package.
 
 ## History
 

ibmmq/mqi.go

@@ -51,11 +51,13 @@ package ibmmq
 #include <string.h>
 #include <cmqc.h>
 #include <cmqxc.h>
+
 */
 import "C"
 
 import (
 	"encoding/binary"
+	"io"
 	"strings"
 	"unsafe"
 )
@@ -119,6 +121,8 @@ func (e *MQReturn) Error() string {
 	return mqstrerror(e.verb, C.MQLONG(e.MQCC), C.MQLONG(e.MQRC))
 }
 
+var endian binary.ByteOrder // Used by structure formatters such as MQCFH
+
 /*
  * Copy a Go string in "strings"
  * to a fixed-size C char array such as MQCHAR12
@@ -138,7 +142,7 @@ func setMQIString(a *C.char, v string, l int) {
 /*
  * The C.GoStringN function can return strings that include
  * NUL characters (which is not really what is expected for a C string-related
- * function). So we have a utility function to remove any trailing nulls
+ * function). So we have a utility function to remove any trailing nulls and spaces
  */
 func trimStringN(c *C.char, l C.int) string {
 	var rc string
@@ -149,7 +153,7 @@ func trimStringN(c *C.char, l C.int) string {
 	} else {
 		rc = s[0:i]
 	}
-	return rc
+	return strings.TrimSpace(rc)
 }
 
 /*
@@ -219,6 +223,7 @@ func (x *MQQueueManager) Disc() error {
 	var mqrc C.MQLONG
 	var mqcc C.MQLONG
 
+	savedConn := x.hConn
 	C.MQDISC(&x.hConn, &mqcc, &mqrc)
 
 	mqreturn := MQReturn{MQCC: int32(mqcc),
@@ -230,6 +235,8 @@ func (x *MQQueueManager) Disc() error {
 		return &mqreturn
 	}
 
+	cbRemoveConnection(savedConn)
+
 	return nil
 }
 
@@ -248,7 +255,7 @@ func (x *MQQueueManager) Open(good *MQOD, goOpenOptions int32) (MQObject, error)
 	}
 
 	copyODtoC(&mqod, good)
-	mqOpenOptions = C.MQLONG(goOpenOptions)
+	mqOpenOptions = C.MQLONG(goOpenOptions) | C.MQOO_FAIL_IF_QUIESCING
 
 	C.MQOPEN(x.hConn,
 		(C.PMQVOID)(unsafe.Pointer(&mqod)),
@@ -288,6 +295,9 @@ func (object *MQObject) Close(goCloseOptions int32) error {
 
 	mqCloseOptions = C.MQLONG(goCloseOptions)
 
+	savedHConn := object.qMgr.hConn
+	savedHObj := object.hObj
+
 	C.MQCLOSE(object.qMgr.hConn, &object.hObj, mqCloseOptions, &mqcc, &mqrc)
 
 	mqreturn := MQReturn{MQCC: int32(mqcc),
@@ -299,6 +309,7 @@ func (object *MQObject) Close(goCloseOptions int32) error {
 		return &mqreturn
 	}
 
+	cbRemoveHandle(savedHConn, savedHObj)
 	return nil
 
 }
@@ -373,6 +384,33 @@ func (subObject *MQObject) Subrq(gosro *MQSRO, action int32) error {
 	return nil
 }
 
+/*
+Begin is the function to start a two-phase XA transaction coordinated by MQ
+*/
+func (x *MQQueueManager) Begin(gobo *MQBO) error {
+	var mqrc C.MQLONG
+	var mqcc C.MQLONG
+	var mqbo C.MQBO
+
+	copyBOtoC(&mqbo, gobo)
+
+	C.MQBEGIN(x.hConn, (C.PMQVOID)(unsafe.Pointer(&mqbo)), &mqcc, &mqrc)
+
+	copyBOfromC(&mqbo, gobo)
+
+	mqreturn := MQReturn{MQCC: int32(mqcc),
+		MQRC: int32(mqrc),
+		verb: "MQBEGIN",
+	}
+
+	if mqcc != C.MQCC_OK {
+		return &mqreturn
+	}
+
+	return nil
+
+}
+
 /*
 Cmit is the function to commit an in-flight transaction
 */
@@ -1214,15 +1252,42 @@ func (handle *MQMessageHandle) InqMP(goimpo *MQIMPO, gopd *MQPD, name string) (s
 			propertyValue = true
 		}
 	case C.MQTYPE_STRING:
-		propertyValue = C.GoStringN((*C.char)(propertyPtr), propertyLength)
+		propertyValue = C.GoStringN((*C.char)(propertyPtr), (C.int)(propertyLength))
 	case C.MQTYPE_BYTE_STRING:
 		ba := make([]byte, propertyLength)
 		p := (*C.MQBYTE)(propertyPtr)
-		copy(ba[:], C.GoBytes(unsafe.Pointer(p), propertyLength))
+		copy(ba[:], C.GoBytes(unsafe.Pointer(p), (C.int)(propertyLength)))
 		propertyValue = ba
 	case C.MQTYPE_NULL:
 		propertyValue = nil
 	}
 
 	return goimpo.ReturnedName, propertyValue, nil
 }
+
+/*
+GetHeader returns a structure containing a parsed-out version of an MQI
+message header such as the MQDLH (which is currently the only structure
+supported). Other structures like the RFH2 could follow.
+
+The caller of this function needs to cast the returned structure to the
+specific type in order to reference the fields.
+*/
+func GetHeader(md *MQMD, buf []byte) (interface{}, int, error) {
+	switch md.Format {
+	case MQFMT_DEAD_LETTER_HEADER:
+		return getHeaderDLH(md, buf)
+	}
+
+	mqreturn := &MQReturn{MQCC: int32(MQCC_FAILED),
+		MQRC: int32(MQRC_FORMAT_NOT_SUPPORTED),
+	}
+
+	return nil, 0, mqreturn
+}
+
+func readStringFromFixedBuffer(r io.Reader, l int32) string {
+	tmpBuf := make([]byte, l)
+	binary.Read(r, endian, tmpBuf)
+	return strings.TrimSpace(string(tmpBuf))
+}

ibmmq/mqiBO.go

@@ -0,0 +1,54 @@
+package ibmmq
+
+/*
+  Copyright (c) IBM Corporation 2018
+
+  Licensed under the Apache License, Version 2.0 (the "License");
+  you may not use this file except in compliance with the License.
+  You may obtain a copy of the License at
+
+  http://www.apache.org/licenses/LICENSE-2.0
+
+  Unless required by applicable law or agreed to in writing, software
+  distributed under the License is distributed on an "AS IS" BASIS,
+  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  See the License for the specific language governing permissions and
+  limitations under the License.
+
+   Contributors:
+     Mark Taylor - Initial Contribution
+*/
+
+/*
+
+#include <stdlib.h>
+#include <string.h>
+#include <cmqc.h>
+
+*/
+import "C"
+
+/*
+This module contains the Begin Options structure
+*/
+
+type MQBO struct {
+	Options int32
+}
+
+func NewMQBO() *MQBO {
+	bo := new(MQBO)
+	bo.Options = int32(C.MQBO_NONE)
+	return bo
+}
+
+func copyBOtoC(mqbo *C.MQBO, gobo *MQBO) {
+	setMQIString((*C.char)(&mqbo.StrucId[0]), "BO  ", 4)
+	mqbo.Version = 1
+	mqbo.Options = C.MQLONG(gobo.Options)
+}
+
+func copyBOfromC(mqbo *C.MQBO, gobo *MQBO) {
+	gobo.Options = int32(mqbo.Options)
+	return
+}

ibmmq/mqiCBC.go

@@ -0,0 +1,78 @@
+package ibmmq
+
+/*
+  Copyright (c) IBM Corporation 2018
+
+  Licensed under the Apache License, Version 2.0 (the "License");
+  you may not use this file except in compliance with the License.
+  You may obtain a copy of the License at
+
+  http://www.apache.org/licenses/LICENSE-2.0
+
+  Unless required by applicable law or agreed to in writing, software
+  distributed under the License is distributed on an "AS IS" BASIS,
+  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+  See the License for the specific language governing permissions and
+  limitations under the License.
+
+   Contributors:
+     Mark Taylor - Initial Contribution
+*/
+
+/*
+
+#include <stdlib.h>
+#include <string.h>
+#include <cmqc.h>
+
+*/
+import "C"
+
+/*
+MQCBC is a structure containing the MQ Callback Context
+The CompCode and Reason in the C structure are not included here. They
+are set in an independent MQReturn structure passed to the callback. Similarly
+for the hObj
+*/
+type MQCBC struct {
+	CallType       int32
+	CallbackArea   interface{} // These fields are saved/restored in parent function
+	ConnectionArea interface{}
+	State          int32
+	DataLength     int32
+	BufferLength   int32
+	Flags          int32
+	ReconnectDelay int32
+}
+
+/*
+NewMQCBC creates a MQCBC structure. There are no default values
+as the structure is created within MQ.
+*/
+func NewMQCBC() *MQCBC {
+	cbc := new(MQCBC)
+	return cbc
+}
+
+/*
+Since we do not create the structure, there's no conversion for it into
+a C format
+*/
+func copyCBCtoC(mqcbc *C.MQCBC, gocbc *MQCBC) {
+	return
+}
+
+/*
+But we do need a conversion process from C
+*/
+func copyCBCfromC(mqcbc *C.MQCBC, gocbc *MQCBC) {
+	gocbc.CallType = int32(mqcbc.CallType)
+	gocbc.State = int32(mqcbc.State)
+	gocbc.DataLength = int32(mqcbc.DataLength)
+	gocbc.BufferLength = int32(mqcbc.BufferLength)
+	gocbc.Flags = int32(mqcbc.Flags)
+	gocbc.ReconnectDelay = int32(mqcbc.ReconnectDelay)
+	// ConnectionArea and CallbackArea are restored outside this function
+
+	return
+}