Merge pull request #146 from remicorniere/PubSub_Example

Pub sub example update

_examples/xmpp_pubsub_client/xmpp_ps_client.go

@@ -32,10 +32,10 @@ func main() {
 	router.NewRoute().Packet("message").
 		HandlerFunc(func(s xmpp.Sender, p stanza.Packet) {
 			data, _ := xml.Marshal(p)
-			fmt.Println("Received a publication ! => \n" + string(data))
+			log.Println("Received a message ! => \n" + string(data))
 		})
 
-	client, err := xmpp.NewClient(config, router, func(err error) { fmt.Println(err) })
+	client, err := xmpp.NewClient(config, router, func(err error) { log.Println(err) })
 	if err != nil {
 		log.Fatalf("%+v", err)
 	}
@@ -49,11 +49,45 @@ func main() {
 
 	// ==========================
 	// Create a node
+	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
+	createNode(ctx, cancel, client)
+
+	// =============================
+	// Configure the node. This can also be done in a single message with the creation
+	configureNode(ctx, cancel, client)
+
+	// ====================================
+	// Subscribe to this node :
+	subToNode(ctx, cancel, client)
+
+	// ==========================
+	// Publish to that node
+	pubToNode(ctx, cancel, client)
+
+	// =============================
+	// Let's purge the node :
+	purgeRq, _ := stanza.NewPurgeAllItems(serviceName, nodeName)
+	purgeCh, err := client.SendIQ(ctx, purgeRq)
+	select {
+	case purgeResp := <-purgeCh:
+		if purgeResp.Error != nil {
+			cancel()
+			log.Fatalf("error while purging node : %s", purgeResp.Error.Text)
+		}
+		log.Println("node successfully purged")
+	case <-time.After(1000 * time.Millisecond):
+		cancel()
+		log.Fatal("No iq response was received in time while purging node")
+	}
+
+	cancel()
+}
+
+func createNode(ctx context.Context, cancel context.CancelFunc, client *xmpp.Client) {
 	rqCreate, err := stanza.NewCreateNode(serviceName, nodeName)
 	if err != nil {
 		log.Fatalf("%+v", err)
 	}
-	ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
 	createCh, err := client.SendIQ(ctx, rqCreate)
 	if err != nil {
 		log.Fatalf("%+v", err)
@@ -67,20 +101,73 @@ func main() {
 					if respCr.Error.Reason != "conflict" {
 						log.Fatalf("%+v", respCr.Error.Text)
 					}
-					fmt.Println(respCr.Error.Text)
+					log.Println(respCr.Error.Text)
 				} else {
 					fmt.Print("successfully created channel")
 				}
 			case <-time.After(100 * time.Millisecond):
 				cancel()
-				log.Fatal("No iq response was received in time")
+				log.Fatal("No iq response was received in time while creating node")
 			}
 		}
 	}
+}
+
+func configureNode(ctx context.Context, cancel context.CancelFunc, client *xmpp.Client) {
+	// First, ask for a form with the config options
+	confRq, _ := stanza.NewConfigureNode(serviceName, nodeName)
+	confReqCh, err := client.SendIQ(ctx, confRq)
+	if err != nil {
+		log.Fatalf("could not send iq : %v", err)
+	}
+	select {
+	case confForm := <-confReqCh:
+		// If the request was successful, we now have a form with configuration options to update
+		fields, err := confForm.GetFormFields()
+		if err != nil {
+			log.Fatal("No config fields found !")
+		}
 
-	// ====================================
+		// These are some common fields expected to be present. Change processing to your liking
-	// Now let's subscribe to this node :
+		if fields["pubsub#max_payload_size"] != nil {
-	rqSubscribe, _ := stanza.NewSubRq(serviceName, stanza.SubInfo{
+			fields["pubsub#max_payload_size"].ValuesList[0] = "100000"
+		}
+
+		if fields["pubsub#notification_type"] != nil {
+			fields["pubsub#notification_type"].ValuesList[0] = "headline"
+		}
+
+		// Send the modified fields as a form
+		submitConf, err := stanza.NewFormSubmissionOwner(serviceName,
+			nodeName,
+			[]*stanza.Field{
+				fields["pubsub#max_payload_size"],
+				fields["pubsub#notification_type"],
+			})
+
+		c, _ := client.SendIQ(ctx, submitConf)
+		select {
+		case confResp := <-c:
+			if confResp.Error != nil {
+				cancel()
+				log.Fatalf("node configuration failed : %s", confResp.Error.Text)
+			}
+			log.Println("node configuration was successful")
+			return
+
+		case <-time.After(300 * time.Millisecond):
+			cancel()
+			log.Fatal("No iq response was received in time while configuring the node")
+		}
+
+	case <-time.After(300 * time.Millisecond):
+		cancel()
+		log.Fatal("No iq response was received in time while asking for the config form")
+	}
+}
+
+func subToNode(ctx context.Context, cancel context.CancelFunc, client *xmpp.Client) {
+	rqSubscribe, err := stanza.NewSubRq(serviceName, stanza.SubInfo{
 		Node: nodeName,
 		Jid:  userJID,
 	})
@@ -91,15 +178,15 @@ func main() {
 	if subRespCh != nil {
 		select {
 		case <-subRespCh:
-			fmt.Println("Subscribed to the service")
+			log.Println("Subscribed to the service")
-		case <-time.After(100 * time.Millisecond):
+		case <-time.After(300 * time.Millisecond):
 			cancel()
-			log.Fatal("No iq response was received in time")
+			log.Fatal("No iq response was received in time while subscribing")
 		}
 	}
+}
 
-	// ==========================
+func pubToNode(ctx context.Context, cancel context.CancelFunc, client *xmpp.Client) {
-	// Publish to that node
 	pub, err := stanza.NewPublishItemRq(serviceName, nodeName, "", stanza.Item{
 		Publisher: "testuser2",
 		Any: &stanza.Node{
@@ -166,17 +253,10 @@ func main() {
 	if pubRespCh != nil {
 		select {
 		case <-pubRespCh:
-			fmt.Println("Published item to the service")
+			log.Println("Published item to the service")
-		case <-time.After(100 * time.Millisecond):
+		case <-time.After(300 * time.Millisecond):
 			cancel()
-			log.Fatal("No iq response was received in time")
+			log.Fatal("No iq response was received in time while publishing")
 		}
 	}
-
-	// =============================
-	// Let's purge the node now :
-	purgeRq, _ := stanza.NewPurgeAllItems(serviceName, nodeName)
-	client.SendIQ(ctx, purgeRq)
-
-	cancel()
 }

stanza/form.go

@@ -17,7 +17,7 @@ type Form struct {
 	XMLName      xml.Name  `xml:"jabber:x:data x"`
 	Instructions []string  `xml:"instructions"`
 	Title        string    `xml:"title,omitempty"`
-	Fields       []Field   `xml:"field,omitempty"`
+	Fields       []*Field  `xml:"field,omitempty"`
 	Reported     *FormItem `xml:"reported"`
 	Items        []FormItem
 	Type         string `xml:"type,attr"`
@@ -38,7 +38,7 @@ type Field struct {
 	Label       string   `xml:"label,attr,omitempty"`
 }
 
-func NewForm(fields []Field, formType string) *Form {
+func NewForm(fields []*Field, formType string) *Form {
 	return &Form{
 		Type:   formType,
 		Fields: fields,

stanza/form_test.go

@@ -57,7 +57,7 @@ func TestMarshalFormSubmit(t *testing.T) {
 			Node: serviceNode,
 			Form: &Form{
 				Type: FormTypeSubmit,
-				Fields: []Field{
+				Fields: []*Field{
 					{Var: "FORM_TYPE", Type: FieldTypeHidden, ValuesList: []string{"http://jabber.org/protocol/pubsub#node_config"}},
 					{Var: "pubsub#title", ValuesList: []string{"Princely Musings (Atom)"}},
 					{Var: "pubsub#deliver_notifications", ValuesList: []string{"1"}},

stanza/pubsub_owner.go

@@ -198,7 +198,7 @@ func NewApprovePendingSubRequest(serviceId, sessionId, nodeId string) (IQ, error
 
 	form := &Form{
 		Type:   FormTypeSubmit,
-		Fields: []Field{{Var: "pubsub#node", ValuesList: []string{nodeId}}},
+		Fields: []*Field{{Var: "pubsub#node", ValuesList: []string{nodeId}}},
 	}
 	data, err := xml.Marshal(form)
 	if err != nil {
@@ -262,25 +262,44 @@ func NewAffiliationListRequest(serviceId, nodeID string) (IQ, error) {
 	return iq, nil
 }
 
+// NewFormSubmission builds a form submission pubsub IQ, in the Owner namespace
+// This is typically used to respond to a form issued by the server when configuring a node.
+// See 8.2.4 Form Submission
+func NewFormSubmissionOwner(serviceId, nodeName string, fields []*Field) (IQ, error) {
+	if serviceId == "" || nodeName == "" {
+		return IQ{}, errors.New("serviceId and nodeName must be filled for this request to be valid")
+	}
+
+	submitConf := NewIQ(Attrs{Type: IQTypeSet, To: serviceId})
+	submitConf.Payload = &PubSubOwner{
+		OwnerUseCase: &ConfigureOwner{
+			Node: nodeName,
+			Form: NewForm(fields,
+				FormTypeSubmit)},
+	}
+
+	return submitConf, nil
+}
+
 // GetFormFields gets the fields from a form in a IQ stanza of type result, as a map.
 // Key is the "var" attribute of the field, and field is the value.
 // The user can then select and modify the fields they want to alter, and submit a new form to the service using the
 // NewFormSubmission function to build the IQ.
 // TODO : remove restriction on IQ type ?
-func (iq *IQ) GetFormFields() (map[string]Field, error) {
+func (iq *IQ) GetFormFields() (map[string]*Field, error) {
 	if iq.Type != IQTypeResult {
 		return nil, errors.New("this IQ is not a result type IQ. Cannot extract the form from it")
 	}
 	switch payload := iq.Payload.(type) {
 	// We support IOT Control IQ
 	case *PubSubGeneric:
-		fieldMap := make(map[string]Field)
+		fieldMap := make(map[string]*Field)
 		for _, elt := range payload.Configure.Form.Fields {
 			fieldMap[elt.Var] = elt
 		}
 		return fieldMap, nil
 	case *PubSubOwner:
-		fieldMap := make(map[string]Field)
+		fieldMap := make(map[string]*Field)
 		co, ok := payload.OwnerUseCase.(*ConfigureOwner)
 		if !ok {
 			return nil, errors.New("this IQ does not contain a PubSub payload with a configure tag for the owner namespace")
@@ -291,7 +310,7 @@ func (iq *IQ) GetFormFields() (map[string]Field, error) {
 		return fieldMap, nil
 	default:
 		if iq.Any != nil {
-			fieldMap := make(map[string]Field)
+			fieldMap := make(map[string]*Field)
 			if iq.Any.XMLName.Local != "command" {
 				return nil, errors.New("this IQ does not contain a form")
 			}
@@ -307,7 +326,7 @@ func (iq *IQ) GetFormFields() (map[string]Field, error) {
 							}
 							err = xml.Unmarshal(data, &f)
 							if err == nil {
-								fieldMap[f.Var] = f
+								fieldMap[f.Var] = &f
 							}
 						}
 					}

stanza/pubsub_owner_test.go

@@ -357,7 +357,7 @@ func TestNewApproveSubRequest(t *testing.T) {
 
 	apprForm := &stanza.Form{
 		Type: stanza.FormTypeSubmit,
-		Fields: []stanza.Field{
+		Fields: []*stanza.Field{
 			{Var: "FORM_TYPE", Type: stanza.FieldTypeHidden, ValuesList: []string{"http://jabber.org/protocol/pubsub#subscribe_authorization"}},
 			{Var: "pubsub#subid", ValuesList: []string{"123-abc"}},
 			{Var: "pubsub#node", ValuesList: []string{"princely_musings"}},
@@ -381,7 +381,7 @@ func TestNewApproveSubRequest(t *testing.T) {
 
 	for _, f := range frm.Fields {
 		if f.Var == "pubsub#allow" {
-			allowField = &f
+			allowField = f
 		}
 	}
 	if allowField == nil || allowField.ValuesList[0] != "true" {
@@ -816,6 +816,58 @@ func TestGetFormFieldsCmd(t *testing.T) {
 
 }
 
+func TestNewFormSubmissionOwner(t *testing.T) {
+	expectedReq := "<iq type=\"set\" id=\"config2\" to=\"pubsub.shakespeare.lit\">" +
+		"<pubsub xmlns=\"http://jabber.org/protocol/pubsub#owner\"> <configure node=\"princely_musings\"> " +
+		"<x xmlns=\"jabber:x:data\" type=\"submit\" > <field var=\"FORM_TYPE\" type=\"hidden\"> " +
+		"<value>http://jabber.org/protocol/pubsub#node_config</value> </field> <field var=\"pubsub#item_expire\"> " +
+		"<value>604800</value> </field> <field var=\"pubsub#access_model\"> <value>roster</value> </field> " +
+		"<field var=\"pubsub#roster_groups_allowed\"> <value>friends</value> <value>servants</value> " +
+		"<value>courtiers</value> </field> </x> </configure> </pubsub> </iq>"
+
+	subR, err := stanza.NewFormSubmissionOwner("pubsub.shakespeare.lit",
+		"princely_musings",
+		[]*stanza.Field{
+			{Var: "FORM_TYPE", Type: stanza.FieldTypeHidden, ValuesList: []string{"http://jabber.org/protocol/pubsub#node_config"}},
+			{Var: "pubsub#item_expire", ValuesList: []string{"604800"}},
+			{Var: "pubsub#access_model", ValuesList: []string{"roster"}},
+			{Var: "pubsub#roster_groups_allowed", ValuesList: []string{"friends", "servants", "courtiers"}},
+		})
+	subR.Id = "config2"
+	if err != nil {
+		t.Fatalf("Could not create request : %s", err)
+	}
+
+	if _, e := checkMarshalling(t, subR); e != nil {
+		t.Fatalf("Failed to check marshalling for generated sub request : %s", e)
+	}
+
+	pubsub, ok := subR.Payload.(*stanza.PubSubOwner)
+	if !ok {
+		t.Fatalf("payload is not a pubsub in namespace owner !")
+	}
+
+	conf, ok := pubsub.OwnerUseCase.(*stanza.ConfigureOwner)
+	if !ok {
+		t.Fatalf("pubsub does not contain a configure node !")
+	}
+
+	if conf.Form == nil {
+		t.Fatalf("the form is absent from the configuration submission !")
+	}
+	if len(conf.Form.Fields) != 4 {
+		t.Fatalf("expected 4 fields, found %d", len(conf.Form.Fields))
+	}
+	if len(conf.Form.Fields[3].ValuesList) != 3 {
+		t.Fatalf("expected 3 values in fourth field, found %d", len(conf.Form.Fields[3].ValuesList))
+	}
+
+	data, err := xml.Marshal(subR)
+	if err := compareMarshal(expectedReq, string(data)); err != nil {
+		t.Fatalf(err.Error())
+	}
+}
+
 func getPubSubOwnerPayload(response string) (*stanza.PubSubOwner, error) {
 	var respIQ stanza.IQ
 	err := xml.Unmarshal([]byte(response), &respIQ)

stanza/pubsub_test.go

@@ -8,7 +8,7 @@ import (
 	"testing"
 )
 
-var submitFormExample = stanza.NewForm([]stanza.Field{
+var submitFormExample = stanza.NewForm([]*stanza.Field{
 	{Var: "FORM_TYPE", Type: stanza.FieldTypeHidden, ValuesList: []string{"http://jabber.org/protocol/pubsub#node_config"}},
 	{Var: "pubsub#title", ValuesList: []string{"Princely Musings (Atom)"}},
 	{Var: "pubsub#deliver_notifications", ValuesList: []string{"1"}},
@@ -741,7 +741,7 @@ func TestNewCreateAndConfigNode(t *testing.T) {
 		"princely_musings",
 		&stanza.Form{
 			Type: stanza.FormTypeSubmit,
-			Fields: []stanza.Field{
+			Fields: []*stanza.Field{
 				{Var: "FORM_TYPE", Type: stanza.FieldTypeHidden, ValuesList: []string{"http://jabber.org/protocol/pubsub#node_config"}},
 				{Var: "pubsub#notify_retract", ValuesList: []string{"0"}},
 				{Var: "pubsub#notify_sub", ValuesList: []string{"0"}},