Merge lp:~mandel/ciborium/remount-drives-take-2 into lp:ciborium

[Manuel de la Peña]

Ensure that we remount the drives after we format them. Remove deadlocks.

The follwing MR does a number of changes to ensure that we can handle the dbus events in an asycn way to work around this go dbus bug https://bugs.launchpad.net/go-dbus/+bug/1416352

* In the code now jobs are handled by a job manager that keeps track of them to later fwd to a channel with their state.
* The udisks code listens to the diff channels and reacts according to them:
    * Additions: new interfaces are added and therefore a new drive is present.
    * Removals: Interfaces are done. A drive has gone.
    * Jobs: the diff jons performed on udisk that do not represent drives (Erase, Format)
* Udisks tracks the diff formatted drives and will mount them again when the Filesystem interface is re-added.
* Tests have been added to the changes.

[lp:~mandel/ciborium/remount-drives-take-2]

93. By Manuel de la Peña on 2015-02-05

Apply go fmt.

94. By Manuel de la Peña on 2015-02-05

Remove ugly log line.

[PS Jenkins bot (ps-jenkins)]

PASSED: Continuous integration, rev:94
http://jenkins.qa.ubuntu.com/job/ciborium-ci/12/
Executed test runs:
    SUCCESS: http://jenkins.qa.ubuntu.com/job/ciborium-vivid-amd64-ci/6
    SUCCESS: http://jenkins.qa.ubuntu.com/job/ciborium-vivid-armhf-ci/6
        deb: http://jenkins.qa.ubuntu.com/job/ciborium-vivid-armhf-ci/6/artifact/work/output/*zip*/output.zip
    SUCCESS: http://jenkins.qa.ubuntu.com/job/ciborium-vivid-i386-ci/6

Click here to trigger a rebuild:
http://s-jenkins.ubuntu-ci:8080/job/ciborium-ci/12/rebuild

review: Approve (continuous-integration)

[Sergio Schvezov (sergiusens)]

Thanks, the code looks good; just a couple of minor comments.

review: Needs Fixing

[Manuel de la Peña (mandel)]

Some answers inline, other changes will come in the next push.

[lp:~mandel/ciborium/remount-drives-take-2]

95. By Manuel de la Peña on 2015-02-09

Update code according to review.

[PS Jenkins bot (ps-jenkins)]

PASSED: Continuous integration, rev:95
http://jenkins.qa.ubuntu.com/job/ciborium-ci/16/
Executed test runs:
    SUCCESS: http://jenkins.qa.ubuntu.com/job/ciborium-vivid-amd64-ci/10
    SUCCESS: http://jenkins.qa.ubuntu.com/job/ciborium-vivid-armhf-ci/10
        deb: http://jenkins.qa.ubuntu.com/job/ciborium-vivid-armhf-ci/10/artifact/work/output/*zip*/output.zip
    SUCCESS: http://jenkins.qa.ubuntu.com/job/ciborium-vivid-i386-ci/10

Click here to trigger a rebuild:
http://s-jenkins.ubuntu-ci:8080/job/ciborium-ci/16/rebuild

review: Approve (continuous-integration)

[Sergio Schvezov (sergiusens)]

Some comments on the fixes (redaction)

and some new copyright header comments.

review: Needs Fixing

[lp:~mandel/ciborium/remount-drives-take-2]

96. By Manuel de la Peña on 2015-02-11

Update docs and license.

97. By Manuel de la Peña on 2015-02-11

More typos.

[PS Jenkins bot (ps-jenkins)]

PASSED: Continuous integration, rev:97
http://jenkins.qa.ubuntu.com/job/ciborium-ci/18/
Executed test runs:
    SUCCESS: http://jenkins.qa.ubuntu.com/job/ciborium-vivid-amd64-ci/12
    SUCCESS: http://jenkins.qa.ubuntu.com/job/ciborium-vivid-armhf-ci/12
        deb: http://jenkins.qa.ubuntu.com/job/ciborium-vivid-armhf-ci/12/artifact/work/output/*zip*/output.zip
    SUCCESS: http://jenkins.qa.ubuntu.com/job/ciborium-vivid-i386-ci/12

Click here to trigger a rebuild:
http://s-jenkins.ubuntu-ci:8080/job/ciborium-ci/18/rebuild

review: Approve (continuous-integration)

[Sergio Schvezov (sergiusens)]

Tested, seems to work fine; there are some other corner cases that need fixing though (format when device is busy).

review: Approve

=== modified file 'cmd/ciborium/main.go'
--- cmd/ciborium/main.go	2014-11-30 23:42:50 +0000
+++ cmd/ciborium/main.go	2015-02-11 15:31:38 +0000
@@ -108,6 +108,8 @@
 }
 
 func main() {
+	// set default logger flags to get more useful info
+	log.SetFlags(log.LstdFlags | log.Lshortfile)
 
 	// Initialize i18n
 	gettext.SetLocale(gettext.LC_ALL, "")

=== added file 'udisks2/common_test.go'
--- udisks2/common_test.go	1970-01-01 00:00:00 +0000
+++ udisks2/common_test.go	2015-02-11 15:31:38 +0000
@@ -0,0 +1,28 @@
+/*
+ * Copyright 2015 Canonical Ltd.
+ *
+ * Authors:
+ * Manuel de la Pena : [e-mail]
+ *
+ * ciborium is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation; version 3.
+ *
+ * ciborium is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+package udisks2
+
+import (
+	"testing"
+
+	. "launchpad.net/gocheck"
+)
+
+func Test(t *testing.T) { TestingT(t) }

=== added file 'udisks2/dispatcher.go'
--- udisks2/dispatcher.go	1970-01-01 00:00:00 +0000
+++ udisks2/dispatcher.go	2015-02-11 15:31:38 +0000
@@ -0,0 +1,174 @@
+/*
+ * Copyright 2015 Canonical Ltd.
+ *
+ * Authors:
+ * Manuel de la Pena : [e-mail]
+ *
+ * ciborium is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation; version 3.
+ *
+ * ciborium is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+package udisks2
+
+import (
+	"log"
+	"runtime"
+	"sort"
+	"strings"
+
+	"launchpad.net/go-dbus/v1"
+)
+
+const (
+	jobPrefixPath    = "/org/freedesktop/UDisks2/jobs/"
+	blockDevicesPath = "/org/freedesktop/UDisks2/block_devices/"
+)
+
+type Interfaces []string
+
+type Event struct {
+	Path       dbus.ObjectPath
+	Props      InterfacesAndProperties
+	Interfaces Interfaces
+}
+
+// isRemovalEvent returns if an event represents an InterfacesRemoved signal from the dbus ObjectManager
+// dbus interface. An event is a removal event when it carries a set of the interfaces that have been lost
+// in a dbus object path.
+func (e *Event) isRemovalEvent() bool {
+	return len(e.Interfaces) != 0
+}
+
+type dispatcher struct {
+	conn           *dbus.Connection
+	additionsWatch *dbus.SignalWatch
+	removalsWatch  *dbus.SignalWatch
+	Jobs           chan Event
+	Additions      chan Event
+	Removals       chan Event
+}
+
+func connectToSignal(conn *dbus.Connection, path dbus.ObjectPath, inter, member string) (*dbus.SignalWatch, error) {
+	log.Println("Connecting to signal", path, inter, member)
+	w, err := conn.WatchSignal(&dbus.MatchRule{
+		Type:      dbus.TypeSignal,
+		Sender:    dbusName,
+		Interface: dbusObjectManagerInterface,
+		Member:    member,
+		Path:      path})
+	return w, err
+}
+
+// newDispatcher tries to return a dispatcher instance that is connected to the dbus signal that must be listened
+// in order to interact with UDisk. If the connection with the signals could not be performed an error is returned.
+func newDispatcher(conn *dbus.Connection) (*dispatcher, error) {
+	log.Print("Creating new dispatcher.")
+	add_w, err := connectToSignal(conn, dbusObject, dbusObjectManagerInterface, dbusAddedSignal)
+	if err != nil {
+		return nil, err
+	}
+
+	remove_w, err := connectToSignal(conn, dbusObject, dbusObjectManagerInterface, dbusRemovedSignal)
+	if err != nil {
+		return nil, err
+	}
+
+	jobs_ch := make(chan Event)
+	additions_ch := make(chan Event)
+	remove_ch := make(chan Event)
+
+	d := &dispatcher{conn, add_w, remove_w, jobs_ch, additions_ch, remove_ch}
+	runtime.SetFinalizer(d, cleanDispatcherData)
+
+	// create the go routines used to grab the events and dispatch them accordingly
+	return d, nil
+}
+
+func (d *dispatcher) Init() {
+	log.Print("Init the dispatcher.")
+	go func() {
+		for msg := range d.additionsWatch.C {
+			var event Event
+			if err := msg.Args(&event.Path, &event.Props); err != nil {
+				log.Print(err)
+				continue
+			}
+			log.Print("New addition event for path ", event.Path, event.Props)
+			d.processAddition(event)
+		}
+	}()
+
+	go func() {
+		for msg := range d.removalsWatch.C {
+			log.Print("New removal event for path.")
+			var event Event
+			if err := msg.Args(&event.Path, &event.Interfaces); err != nil {
+				log.Print(err)
+				continue
+			}
+			sort.Strings(event.Interfaces)
+			log.Print("Removal event is ", event.Path, " Interfaces: ", event.Interfaces)
+			d.processRemoval(event)
+		}
+	}()
+}
+
+func (d *dispatcher) free() {
+	log.Print("Cleaning dispatcher resources.")
+	// cancel all watches so that goroutines are done and close the
+	// channels
+	d.additionsWatch.Cancel()
+	d.removalsWatch.Cancel()
+	close(d.Jobs)
+	close(d.Additions)
+	close(d.Removals)
+}
+
+func (d *dispatcher) processAddition(event Event) {
+	log.Print("Processing an add event from path ", event.Path)
+	// according to the object path we know if the even was a job one or not
+	if strings.HasPrefix(string(event.Path), jobPrefixPath) {
+		log.Print("Sending a new job event.")
+		select {
+		case d.Jobs <- event:
+			log.Print("Sent event ", event.Path)
+		}
+	} else {
+		log.Print("Sending a new general add event.")
+		select {
+		case d.Additions <- event:
+			log.Print("Sent event ", event.Path)
+		}
+	}
+}
+
+func (d *dispatcher) processRemoval(event Event) {
+	log.Print("Processing a remove event from path ", event.Path)
+	// according to the object path we know if the even was a job one or not
+	if strings.HasPrefix(string(event.Path), jobPrefixPath) {
+		log.Print("Sending a new remove job event.")
+		select {
+		case d.Jobs <- event:
+			log.Println("Sent event", event.Path)
+		}
+	} else {
+		log.Print("Sending a new general remove event.")
+		select {
+		case d.Removals <- event:
+			log.Println("Sent event", event.Path)
+		}
+	}
+}
+
+func cleanDispatcherData(d *dispatcher) {
+	d.free()
+}

=== added file 'udisks2/jobs.go'
--- udisks2/jobs.go	1970-01-01 00:00:00 +0000
+++ udisks2/jobs.go	2015-02-11 15:31:38 +0000
@@ -0,0 +1,147 @@
+/*
+ * Copyright 2015 Canonical Ltd.
+ *
+ * Authors:
+ * Manuel de la Pena : [e-mail]
+ *
+ * ciborium is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation; version 3.
+ *
+ * ciborium is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+package udisks2
+
+import (
+	"log"
+	"runtime"
+	"sort"
+
+	"launchpad.net/go-dbus/v1"
+)
+
+type job struct {
+	Event        Event
+	Operation    string
+	Paths        []string
+	WasCompleted bool
+}
+
+type jobManager struct {
+	onGoingJobs     map[dbus.ObjectPath]job
+	FormatEraseJobs chan job
+	FormatMkfsJobs  chan job
+}
+
+func newJobManager(d *dispatcher) *jobManager {
+	// listen to the diff job events and ensure that they are dealt with in the correct channel
+	ongoing := make(map[dbus.ObjectPath]job)
+	eraseChan := make(chan job)
+	mkfsChan := make(chan job)
+	m := &jobManager{ongoing, eraseChan, mkfsChan}
+	runtime.SetFinalizer(m, cleanJobData)
+
+	// create a go routine that will filter the diff jobs
+	go func() {
+		for {
+			select {
+			case e := <-d.Jobs:
+				log.Println("New event", e.Path, "Properties:", e.Props, "Interfaces:", e.Interfaces)
+				if e.isRemovalEvent() {
+					log.Print("Is removal event")
+					m.processRemovalEvent(e)
+				} else {
+					m.processAdditionEvent(e)
+				}
+			}
+		}
+		log.Print("Job manager routine done")
+	}()
+	return m
+}
+
+func (m *jobManager) processRemovalEvent(e Event) {
+	log.Println("Deal with job event removal", e.Path, e.Interfaces)
+	if job, ok := m.onGoingJobs[e.Path]; ok {
+		// assert that we did loose the jobs interface, the dispatcher does sort the interfaces
+		i := sort.SearchStrings(e.Interfaces, dbusJobInterface)
+		if i != len(e.Interfaces) {
+			log.Print("Job completed.")
+			// complete event found
+			job.WasCompleted = true
+
+			if job.Operation == formatErase {
+				log.Print("Sending completed erase job")
+				m.FormatEraseJobs <- job
+			}
+
+			if job.Operation == formateMkfs {
+				log.Print("Sending completed mkfs job")
+				m.FormatMkfsJobs <- job
+			}
+
+			log.Print("Removed ongoing job for path", e.Path)
+			delete(m.onGoingJobs, e.Path)
+			return
+		} else {
+			log.Println("Ignoring event for path", e.Path, "because the job interface was not lost")
+			return
+		}
+	} else {
+		log.Println("Ignoring event for path", e.Path)
+		return
+	}
+}
+
+func (m *jobManager) processAdditionEvent(e Event) {
+	j, ok := m.onGoingJobs[e.Path]
+	if !ok {
+		log.Println("Creating job for new path", e.Path)
+		log.Println("New job operation", e.Props.jobOperation())
+		operation := e.Props.jobOperation()
+		var paths []string
+		if e.Props.isMkfsFormatJob() {
+			log.Print("Get paths from formatMkfs event.")
+			paths = e.Props.getFormattedPaths()
+		}
+
+		j = job{e, operation, paths, false}
+		m.onGoingJobs[e.Path] = j
+	} else {
+		log.Print("Updating job for path ", e.Path)
+		j.Event = e
+		if e.Props.isEraseFormatJob() {
+			j.Operation = formatErase
+		}
+		if e.Props.isMkfsFormatJob() {
+			j.Operation = formateMkfs
+			j.Paths = e.Props.getFormattedPaths()
+		}
+	}
+
+	if j.Operation == formatErase {
+		log.Print("Sending rease job from addition.")
+		m.FormatEraseJobs <- j
+	} else if j.Operation == formateMkfs {
+		log.Print("Sending format job from addition.")
+		m.FormatMkfsJobs <- j
+	} else {
+		log.Println("Ignoring job event with operation", j.Operation)
+	}
+}
+
+func (m *jobManager) free() {
+	close(m.FormatEraseJobs)
+	close(m.FormatMkfsJobs)
+}
+
+func cleanJobData(m *jobManager) {
+	m.free()
+}

=== added file 'udisks2/jobs_test.go'
--- udisks2/jobs_test.go	1970-01-01 00:00:00 +0000
+++ udisks2/jobs_test.go	2015-02-11 15:31:38 +0000
@@ -0,0 +1,192 @@
+/*
+ * Copyright 2015 Canonical Ltd.
+ *
+ * Authors:
+ * Manuel de la Pena : [e-mail]
+ *
+ * ciborium is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation; version 3.
+ *
+ * ciborium is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+package udisks2
+
+import (
+	"launchpad.net/go-dbus/v1"
+	. "launchpad.net/gocheck"
+)
+
+type JobManagerTestSuite struct {
+	ongoing   map[dbus.ObjectPath]job
+	events    chan Event
+	manager   *jobManager
+	completed chan bool
+}
+
+var _ = Suite(&JobManagerTestSuite{})
+
+func (s *JobManagerTestSuite) SetUpTest(c *C) {
+	s.ongoing = make(map[dbus.ObjectPath]job)
+	eraseChan := make(chan job)
+	mkfsChan := make(chan job)
+
+	s.manager = &jobManager{s.ongoing, eraseChan, mkfsChan}
+	s.completed = make(chan bool)
+}
+
+func (s *JobManagerTestSuite) TearDownTest(c *C) {
+	close(s.manager.FormatEraseJobs)
+	close(s.manager.FormatMkfsJobs)
+	close(s.completed)
+}
+
+func (s *JobManagerTestSuite) TestProcessAddEventNewErase(c *C) {
+	path := dbus.ObjectPath("/org/freedesktop/UDisks2/jobs/3")
+
+	go func() {
+		for j := range s.manager.FormatEraseJobs {
+			c.Assert(j.Operation, Equals, formatErase)
+			c.Assert(j.WasCompleted, Equals, false)
+			// assert that the job is present in the ongoing map
+			_, ok := s.ongoing[path]
+			c.Assert(ok, Equals, true)
+			s.completed <- true
+		}
+	}()
+
+	interfaces := make([]string, 0, 0)
+
+	props := make(map[string]VariantMap)
+	props[dbusJobInterface] = make(map[string]dbus.Variant)
+	props[dbusJobInterface][operationProperty] = dbus.Variant{formatErase}
+	objsPaths := make([]string, 1, 1)
+	objsPaths[0] = "/path/to/erased/fs"
+	props[dbusJobInterface][objectsProperty] = dbus.Variant{objsPaths}
+
+	event := Event{path, props, interfaces}
+	s.manager.processAdditionEvent(event)
+	<-s.completed
+}
+
+func (s *JobManagerTestSuite) TestProcessAddEventNewFormat(c *C) {
+	path := dbus.ObjectPath("/org/freedesktop/UDisks2/jobs/1")
+
+	go func() {
+		for j := range s.manager.FormatMkfsJobs {
+			c.Assert(j.Operation, Equals, formateMkfs)
+			c.Assert(j.WasCompleted, Equals, false)
+			// assert that the job is present in the ongoing map
+			_, ok := s.ongoing[path]
+			c.Assert(ok, Equals, true)
+			s.completed <- true
+		}
+	}()
+
+	interfaces := make([]string, 0, 0)
+
+	props := make(map[string]VariantMap)
+	props[dbusJobInterface] = make(map[string]dbus.Variant)
+	props[dbusJobInterface][operationProperty] = dbus.Variant{formateMkfs}
+	objsPaths := make([]interface{}, 1, 1)
+	objsPaths[0] = "/path/to/new/fs"
+	props[dbusJobInterface][objectsProperty] = dbus.Variant{objsPaths}
+
+	event := Event{path, props, interfaces}
+	s.manager.processAdditionEvent(event)
+	<-s.completed
+}
+
+func (s *JobManagerTestSuite) TestProcessAddEventPresent(c *C) {
+	path := dbus.ObjectPath("/org/freedesktop/UDisks2/jobs/1")
+
+	// add a ongoing job for the given path
+	s.ongoing[path] = job{}
+
+	go func() {
+		for j := range s.manager.FormatMkfsJobs {
+			c.Assert(j.Operation, Equals, formateMkfs)
+			c.Assert(j.WasCompleted, Equals, false)
+			// assert that the job is present in the ongoing map
+			_, ok := s.ongoing[path]
+			c.Assert(ok, Equals, true)
+			s.completed <- true
+		}
+	}()
+
+	interfaces := make([]string, 0, 0)
+
+	props := make(map[string]VariantMap)
+	props[dbusJobInterface] = make(map[string]dbus.Variant)
+	props[dbusJobInterface][operationProperty] = dbus.Variant{formateMkfs}
+	objsPaths := make([]interface{}, 1, 1)
+	objsPaths[0] = "/path/to/new/fs"
+	props[dbusJobInterface][objectsProperty] = dbus.Variant{objsPaths}
+
+	event := Event{path, props, interfaces}
+	s.manager.processAdditionEvent(event)
+	<-s.completed
+}
+
+func (s *JobManagerTestSuite) TestProcessRemovalEventMissing(c *C) {
+	path := dbus.ObjectPath("/org/freedesktop/UDisks2/jobs/1")
+	interfaces := make([]string, 1, 1)
+	interfaces[0] = dbusJobInterface
+	props := make(map[string]VariantMap)
+
+	event := Event{path, props, interfaces}
+	// nothing bad should happen
+	s.manager.processRemovalEvent(event)
+
+}
+
+func (s *JobManagerTestSuite) TestProcessRemovalEventInterfaceMissing(c *C) {
+	path := dbus.ObjectPath("/org/freedesktop/UDisks2/jobs/1")
+	interfaces := make([]string, 1, 1)
+	interfaces[0] = "com.test.Random"
+	props := make(map[string]VariantMap)
+
+	event := Event{path, props, interfaces}
+
+	s.ongoing[path] = job{}
+
+	// nothing bad should happen
+	s.manager.processRemovalEvent(event)
+}
+
+func (s *JobManagerTestSuite) TestProcessRemovalEventMkfs(c *C) {
+	path := dbus.ObjectPath("/org/freedesktop/UDisks2/jobs/1")
+
+	// create an erase job and add it to the ongoing map, check that the job
+	// is fwd to the channel as completed and removed from the map
+	formattedPaths := make([]string, 1, 1)
+	formattedPaths[0] = "/one/path/to/a/fmormatted/fs/1"
+	presentJob := job{Event{}, formateMkfs, formattedPaths, false}
+
+	s.ongoing[path] = presentJob
+
+	go func() {
+		for j := range s.manager.FormatMkfsJobs {
+			c.Assert(j.Operation, Equals, formateMkfs)
+			c.Assert(j.WasCompleted, Equals, true)
+			c.Assert(len(j.Paths), Equals, 1)
+			s.completed <- true
+		}
+	}()
+
+	interfaces := make([]string, 1, 1)
+	interfaces[0] = dbusJobInterface
+	props := make(map[string]VariantMap)
+
+	event := Event{path, props, interfaces}
+
+	s.manager.processRemovalEvent(event)
+	<-s.completed
+}

=== added file 'udisks2/properties.go'
--- udisks2/properties.go	1970-01-01 00:00:00 +0000
+++ udisks2/properties.go	2015-02-11 15:31:38 +0000
@@ -0,0 +1,145 @@
+/*
+ * Copyright 2015 Canonical Ltd.
+ *
+ * Authors:
+ * Sergio Schvezov: [e-mail]
+ * Manuel de la Pena: [e-mail]
+ *
+ * ciborium is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation; version 3.
+ *
+ * ciborium is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+package udisks2
+
+import (
+	"reflect"
+
+	"launchpad.net/go-dbus/v1"
+)
+
+const (
+	formatErase           = "format-erase"
+	formateMkfs           = "format-mkfs"
+	mountPointsProperty   = "MountPoints"
+	uuidProperty          = "UUID"
+	tableProperty         = "Table"
+	partitionableProperty = "HintPartitionable"
+	operationProperty     = "Operation"
+	objectsProperty       = "Objects"
+)
+
+type VariantMap map[string]dbus.Variant
+type InterfacesAndProperties map[string]VariantMap
+
+func (i InterfacesAndProperties) isMounted() bool {
+	propFS, ok := i[dbusFilesystemInterface]
+	if !ok {
+		return false
+	}
+	mountpointsVariant, ok := propFS[mountPointsProperty]
+	if !ok {
+		return false
+	}
+	if reflect.TypeOf(mountpointsVariant.Value).Kind() != reflect.Slice {
+		return false
+	}
+	mountpoints := reflect.ValueOf(mountpointsVariant.Value).Len()
+
+	return mountpoints > 0
+}
+
+func (i InterfacesAndProperties) hasPartition() bool {
+	prop, ok := i[dbusPartitionInterface]
+	if !ok {
+		return false
+	}
+	// check if a couple of properties exist
+	if _, ok := prop[uuidProperty]; !ok {
+		return false
+	}
+	if _, ok := prop[tableProperty]; !ok {
+		return false
+	}
+	return true
+}
+
+func (i InterfacesAndProperties) isPartitionable() bool {
+	prop, ok := i[dbusBlockInterface]
+	if !ok {
+		return false
+	}
+	partitionableHintVariant, ok := prop[partitionableProperty]
+	if !ok {
+		return false
+	}
+	if reflect.TypeOf(partitionableHintVariant.Value).Kind() != reflect.Bool {
+		return false
+	}
+	return reflect.ValueOf(partitionableHintVariant.Value).Bool()
+}
+
+func (i InterfacesAndProperties) jobOperation() string {
+	prop, ok := i[dbusJobInterface]
+	if !ok {
+		return ""
+	}
+	operationVariant, ok := prop[operationProperty]
+	if !ok {
+		return ""
+	}
+	if reflect.TypeOf(operationVariant.Value).Kind() != reflect.String {
+		return ""
+	}
+	return reflect.ValueOf(operationVariant.Value).String()
+}
+
+func (i InterfacesAndProperties) isEraseFormatJob() bool {
+	return i.jobOperation() == formatErase
+
+}
+
+func (i InterfacesAndProperties) isMkfsFormatJob() bool {
+	return i.jobOperation() == formateMkfs
+}
+
+func (i InterfacesAndProperties) getFormattedPaths() []string {
+	var objectPaths []string
+	prop, ok := i[dbusJobInterface]
+	if !ok {
+		return objectPaths
+	}
+	operationVariant, ok := prop[operationProperty]
+	if !ok {
+		return objectPaths
+	}
+
+	operationStr := reflect.ValueOf(operationVariant.Value).String()
+	if operationStr == formateMkfs {
+		objs, ok := prop[objectsProperty]
+		if ok {
+			objsVal := reflect.ValueOf(objs.Value)
+			length := objsVal.Len()
+			objectPaths = make([]string, length, length)
+			for i := 0; i < length; i++ {
+				objectPaths[i] = objsVal.Index(i).Elem().String()
+			}
+			return objectPaths
+		}
+	}
+
+	return objectPaths
+}
+
+func (i InterfacesAndProperties) isFilesystem() bool {
+	_, ok := i[dbusFilesystemInterface]
+	return ok
+}

=== added file 'udisks2/properties_test.go'
--- udisks2/properties_test.go	1970-01-01 00:00:00 +0000
+++ udisks2/properties_test.go	2015-02-11 15:31:38 +0000
@@ -0,0 +1,210 @@
+/*
+ * Copyright 2015 Canonical Ltd.
+ *
+ * Authors:
+ * Manuel de la Pena : [e-mail]
+ *
+ * ciborium is free software; you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation; version 3.
+ *
+ * ciborium is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+
+package udisks2
+
+import (
+	"sort"
+
+	"launchpad.net/go-dbus/v1"
+	. "launchpad.net/gocheck"
+)
+
+type InterfacesAndPropertiesTestSuite struct {
+	properties InterfacesAndProperties
+}
+
+var _ = Suite(&InterfacesAndPropertiesTestSuite{})
+
+func (s *InterfacesAndPropertiesTestSuite) SetUpTest(c *C) {
+	s.properties = make(map[string]VariantMap)
+}
+
+func (s *InterfacesAndPropertiesTestSuite) TestIsMountedMissingInterface(c *C) {
+	// empty properties means that the interface is missing
+	c.Assert(s.properties.isMounted(), Equals, false)
+}
+
+func (s *InterfacesAndPropertiesTestSuite) TestIsMountedMissingMountPoints(c *C) {
+	// add the expected interface but without the mount points property
+	s.properties[dbusFilesystemInterface] = make(map[string]dbus.Variant)
+	c.Assert(s.properties.isMounted(), Equals, false)
+}
+
+func (s *InterfacesAndPropertiesTestSuite) TestIsMountedNotSlize(c *C) {
+	s.properties[dbusFilesystemInterface] = make(map[string]dbus.Variant)
+	s.properties[dbusFilesystemInterface]["MountPoints"] = dbus.Variant{5}
+	c.Assert(s.properties.isMounted(), Equals, false)
+}
+
+func (s *InterfacesAndPropertiesTestSuite) TestIsMountedZeroMountPoints(c *C) {
+	mount_points := make([]string, 0, 0)
+	s.properties[dbusFilesystemInterface] = make(map[string]dbus.Variant)
+	s.properties[dbusFilesystemInterface]["MountPoints"] = dbus.Variant{mount_points}
+	c.Assert(s.properties.isMounted(), Equals, false)
+}
+
+func (s *InterfacesAndPropertiesTestSuite) TestIsMountedSeveralMountPoints(c *C) {
+	mount_points := make([]string, 1, 1)
+	mount_points[0] = "/random/mount/point"
+	s.properties[dbusFilesystemInterface] = make(map[string]dbus.Variant)
+	s.properties[dbusFilesystemInterface]["MountPoints"] = dbus.Variant{mount_points}
+	c.Assert(s.properties.isMounted(), Equals, true)
+}
+
+func (s *InterfacesAndPropertiesTestSuite) TestHasPartitionMissingInterface(c *C) {
+	// an empty map should result in false
+	c.Assert(s.properties.hasPartition(), Equals, false)
+}
+
+func (s *InterfacesAndPropertiesTestSuite) TestHasPartitionMissinUUID(c *C) {
+	// add the interface with no properties
+	s.properties[dbusPartitionInterface] = make(map[string]dbus.Variant)
+	c.Assert(s.properties.hasPartition(), Equals, false)
+}
+
+func (s *InterfacesAndPropertiesTestSuite) TestHasParitionMissingTable(c *C) {
+	s.properties[dbusPartitionInterface] = make(map[string]dbus.Variant)
+	s.properties[dbusPartitionInterface]["UUID"] = dbus.Variant{"A UUID"}
+	c.Assert(s.properties.hasPartition(), Equals, false)
+}
+
+func (s *InterfacesAndPropertiesTestSuite) TestHasParitionPresent(c *C) {
+	s.properties[dbusPartitionInterface] = make(map[string]dbus.Variant)
+	s.properties[dbusPartitionInterface]["UUID"] = dbus.Variant{"A UUID"}
+	s.properties[dbusPartitionInterface]["Table"] = dbus.Variant{"A Table"}
+	c.Assert(s.properties.hasPartition(), Equals, true)
+}
+
+func (s *InterfacesAndPropertiesTestSuite) TestIsPartitionableMissingInterface(c *C) {
+	// an empty map should result in false
+	c.Assert(s.properties.isPartitionable(), Equals, false)
+}
+
+func (s *InterfacesAndPropertiesTestSuite) TestIsParitionableMissingHint(c *C) {
+	s.properties[dbusBlockInterface] = make(map[string]dbus.Variant)
+	c.Assert(s.properties.isPartitionable(), Equals, false)
+}
+
+func (s *InterfacesAndPropertiesTestSuite) TestIsParitionableHintNotBool(c *C) {
+	s.properties[dbusBlockInterface] = make(map[string]dbus.Variant)
+	s.properties[dbusBlockInterface]["HintPartitionable"] = dbus.Variant{"A String"}
+	c.Assert(s.properties.isPartitionable(), Equals, false)
+}
+
+func (s *InterfacesAndPropertiesTestSuite) TestIsPartitionable(c *C) {
+	s.properties[dbusBlockInterface] = make(map[string]dbus.Variant)
+	s.properties[dbusBlockInterface]["HintPartitionable"] = dbus.Variant{true}
+	c.Assert(s.properties.isPartitionable(), Equals, true)
+}
+
+func (s *InterfacesAndPropertiesTestSuite) TestIsNotPartitionable(c *C) {
+	s.properties[dbusBlockInterface] = make(map[string]dbus.Variant)
+	s.properties[dbusBlockInterface]["HintPartitionable"] = dbus.Variant{false}
+	c.Assert(s.properties.isPartitionable(), Equals, false)
+}
+
+func (s *InterfacesAndPropertiesTestSuite) TestIsEraseFormatJobMissingInterface(c *C) {
+	// an empty map should result in false
+	c.Assert(s.properties.isEraseFormatJob(), Equals, false)
+}
+
+func (s *InterfacesAndPropertiesTestSuite) TestIsEraseFormatJobMissingOperation(c *C) {
+	s.properties[dbusJobInterface] = make(map[string]dbus.Variant)
+	c.Assert(s.properties.isEraseFormatJob(), Equals, false)
+}
+
+func (s *InterfacesAndPropertiesTestSuite) TestIsEraseFormatJobWrongType(c *C) {
+	s.properties[dbusJobInterface] = make(map[string]dbus.Variant)
+	s.properties[dbusJobInterface]["Operation"] = dbus.Variant{false}
+	c.Assert(s.properties.isEraseFormatJob(), Equals, false)
+}
+
+func (s *InterfacesAndPropertiesTestSuite) TestIsEraseFormatJobWrongOperation(c *C) {
+	s.properties[dbusJobInterface] = make(map[string]dbus.Variant)
+	s.properties[dbusJobInterface]["Operation"] = dbus.Variant{"false"}
+	c.Assert(s.properties.isEraseFormatJob(), Equals, false)
+}
+
+func (s *InterfacesAndPropertiesTestSuite) TestIsEraseFormatJob(c *C) {
+	s.properties[dbusJobInterface] = make(map[string]dbus.Variant)
+	s.properties[dbusJobInterface]["Operation"] = dbus.Variant{"format-erase"}
+	c.Assert(s.properties.isEraseFormatJob(), Equals, true)
+}
+
+func (s *InterfacesAndPropertiesTestSuite) TestIsMkfsFormatJobMissingInterface(c *C) {
+	c.Assert(s.properties.isMkfsFormatJob(), Equals, false)
+}
+
+func (s *InterfacesAndPropertiesTestSuite) TestIsMkfsFormatJobMissingOperation(c *C) {
+	s.properties[dbusJobInterface] = make(map[string]dbus.Variant)
+	c.Assert(s.properties.isMkfsFormatJob(), Equals, false)
+}
+
+func (s *InterfacesAndPropertiesTestSuite) TestIsMkfsFormatJobWrongType(c *C) {
+	s.properties[dbusJobInterface] = make(map[string]dbus.Variant)
+	s.properties[dbusJobInterface]["Operation"] = dbus.Variant{true}
+	c.Assert(s.properties.isMkfsFormatJob(), Equals, false)
+}
+
+func (s *InterfacesAndPropertiesTestSuite) TestIsMkfsFormatJobWrongOperation(c *C) {
+	s.properties[dbusJobInterface] = make(map[string]dbus.Variant)
+	s.properties[dbusJobInterface]["Operation"] = dbus.Variant{"false"}
+	c.Assert(s.properties.isMkfsFormatJob(), Equals, false)
+}
+
+func (s *InterfacesAndPropertiesTestSuite) TestIsMkfsFormatJob(c *C) {
+	s.properties[dbusJobInterface] = make(map[string]dbus.Variant)
+	s.properties[dbusJobInterface]["Operation"] = dbus.Variant{"format-mkfs"}
+	c.Assert(s.properties.isMkfsFormatJob(), Equals, true)
+}
+
+func (s *InterfacesAndPropertiesTestSuite) TestGetFormattedPathsMissingInterface(c *C) {
+	paths := s.properties.getFormattedPaths()
+	c.Assert(len(paths), Equals, 0)
+}
+
+func (s *InterfacesAndPropertiesTestSuite) TestGetFormattedPathsMissingProperty(c *C) {
+	s.properties[dbusJobInterface] = make(map[string]dbus.Variant)
+	paths := s.properties.getFormattedPaths()
+	c.Assert(len(paths), Equals, 0)
+}
+
+func (s *InterfacesAndPropertiesTestSuite) TestGetFormattedPaths(c *C) {
+	firstPath := "/path/to/new/fs/1"
+	secondPath := "/path/to/new/fs/2"
+	thirdPath := "/path/to/new/fs/3"
+
+	objsPaths := make([]interface{}, 3, 3)
+	objsPaths[0] = firstPath
+	objsPaths[1] = secondPath
+	objsPaths[2] = thirdPath
+
+	s.properties[dbusJobInterface] = make(map[string]dbus.Variant)
+	s.properties[dbusJobInterface]["Operation"] = dbus.Variant{"format-mkfs"}
+	s.properties[dbusJobInterface]["Objects"] = dbus.Variant{objsPaths}
+
+	paths := s.properties.getFormattedPaths()
+	//sort.Strings(paths)
+
+	c.Assert(len(paths), Equals, len(objsPaths))
+	c.Assert(sort.SearchStrings(paths, firstPath), Not(Equals), len(paths))
+	c.Assert(sort.SearchStrings(paths, secondPath), Not(Equals), len(paths))
+	c.Assert(sort.SearchStrings(paths, thirdPath), Not(Equals), len(paths))
+}

=== modified file 'udisks2/udisks2.go'
--- udisks2/udisks2.go	2014-09-24 23:02:18 +0000
+++ udisks2/udisks2.go	2015-02-11 15:31:38 +0000
@@ -43,16 +43,13 @@
 	dbusFilesystemInterface     = "org.freedesktop.UDisks2.Filesystem"
 	dbusPartitionInterface      = "org.freedesktop.UDisks2.Partition"
 	dbusPartitionTableInterface = "org.freedesktop.UDisks2.PartitionTable"
+	dbusJobInterface            = "org.freedesktop.UDisks2.Job"
 	dbusAddedSignal             = "InterfacesAdded"
 	dbusRemovedSignal           = "InterfacesRemoved"
 )
 
 var ErrUnhandledFileSystem = errors.New("unhandled filesystem")
 
-type VariantMap map[string]dbus.Variant
-type InterfacesAndProperties map[string]VariantMap
-type Interfaces []string
-
 type Drive struct {
 	path         dbus.ObjectPath
 	blockDevices map[dbus.ObjectPath]InterfacesAndProperties
@@ -61,34 +58,33 @@
 
 type driveMap map[dbus.ObjectPath]*Drive
 
-type Event struct {
-	Path  dbus.ObjectPath
-	Props InterfacesAndProperties
-}
-
 type mountpointMap map[dbus.ObjectPath]string
 
 type UDisks2 struct {
-	conn         *dbus.Connection
-	validFS      sort.StringSlice
-	blockAdded   chan *Event
-	driveAdded   *dbus.SignalWatch
-	mountRemoved chan string
-	blockError   chan error
-	driveRemoved *dbus.SignalWatch
-	blockDevice  chan bool
-	drives       driveMap
-	mountpoints  mountpointMap
-	mapLock      sync.Mutex
-	startLock    sync.Mutex
+	conn          *dbus.Connection
+	validFS       sort.StringSlice
+	blockAdded    chan *Event
+	driveAdded    *dbus.SignalWatch
+	mountRemoved  chan string
+	blockError    chan error
+	driveRemoved  *dbus.SignalWatch
+	blockDevice   chan bool
+	drives        driveMap
+	mountpoints   mountpointMap
+	mapLock       sync.Mutex
+	startLock     sync.Mutex
+	dispatcher    *dispatcher
+	jobs          *jobManager
+	pendingMounts []string
 }
 
 func NewStorageWatcher(conn *dbus.Connection, filesystems ...string) (u *UDisks2) {
 	u = &UDisks2{
-		conn:        conn,
-		validFS:     sort.StringSlice(filesystems),
-		drives:      make(driveMap),
-		mountpoints: make(mountpointMap),
+		conn:          conn,
+		validFS:       sort.StringSlice(filesystems),
+		drives:        make(driveMap),
+		mountpoints:   make(mountpointMap),
+		pendingMounts: make([]string, 0, 0),
 	}
 	runtime.SetFinalizer(u, cleanDriveWatch)
 	return u
@@ -212,45 +208,43 @@
 }
 
 func (u *UDisks2) Init() (err error) {
-	if u.driveAdded, err = u.connectToSignalInterfacesAdded(); err != nil {
-		return err
-	}
-	if u.driveRemoved, err = u.connectToSignalInterfacesRemoved(); err != nil {
-		return err
-	}
-	u.initInterfacesWatchChan()
-	return nil
-}
-
-func (u *UDisks2) initInterfacesWatchChan() {
-	go func() {
-		for {
-			select {
-			case msg := <-u.driveAdded.C:
-				var event Event
-				if err := msg.Args(&event.Path, &event.Props); err != nil {
-					log.Print(err)
-					continue
-				}
-				if err := u.processAddEvent(&event); err != nil {
-					log.Print("Issues while processing ", event.Path, ": ", err)
-				}
-			case msg := <-u.driveRemoved.C:
-				var objectPath dbus.ObjectPath
-				var interfaces Interfaces
-				if err := msg.Args(&objectPath, &interfaces); err != nil {
-					log.Print(err)
-					continue
-				}
-				if err := u.processRemoveEvent(objectPath, interfaces); err != nil {
-					log.Println("Issues while processing remove event:", err)
+	d, err := newDispatcher(u.conn)
+	if err == nil {
+		u.dispatcher = d
+		u.jobs = newJobManager(d)
+		go func() {
+			for {
+				select {
+				case e := <-u.dispatcher.Additions:
+					if err := u.processAddEvent(&e); err != nil {
+						log.Print("Issues while processing ", e.Path, ": ", err)
+					}
+				case e := <-u.dispatcher.Removals:
+					if err := u.processRemoveEvent(e.Path, e.Interfaces); err != nil {
+						log.Println("Issues while processing remove event:", err)
+					}
+				case j := <-u.jobs.FormatEraseJobs:
+					if j.WasCompleted {
+						log.Print("Erase job completed.")
+					} else {
+						log.Print("Erase job started.")
+					}
+				case j := <-u.jobs.FormatMkfsJobs:
+					if j.WasCompleted {
+						log.Println("Format job done for", j.Event.Path)
+						u.pendingMounts = append(u.pendingMounts, j.Paths...)
+						sort.Strings(u.pendingMounts)
+					} else {
+						log.Print("Format job started.")
+					}
 				}
 			}
-		}
-		log.Print("Shutting down InterfacesAdded channel")
-	}()
-
-	u.emitExistingDevices()
+		}()
+		d.Init()
+		u.emitExistingDevices()
+		return nil
+	}
+	return err
 }
 
 func (u *UDisks2) connectToSignal(path dbus.ObjectPath, inter, member string) (*dbus.SignalWatch, error) {
@@ -288,7 +282,7 @@
 	var blocks, drives []*Event
 	// separate drives from blocks to avoid aliasing
 	for objectPath, props := range allDevices {
-		s := &Event{objectPath, props}
+		s := &Event{objectPath, props, make([]string, 0, 0)}
 		switch objectPathType(objectPath) {
 		case deviceTypeDrive:
 			drives = append(drives, s)
@@ -313,6 +307,13 @@
 func (u *UDisks2) processAddEvent(s *Event) error {
 	u.mapLock.Lock()
 	defer u.mapLock.Unlock()
+	pos := sort.SearchStrings(u.pendingMounts, string(s.Path))
+	if pos != len(u.pendingMounts) && s.Props.isFilesystem() {
+		log.Println("Mount path", s.Path)
+		_, err := u.Mount(s)
+		u.pendingMounts = append(u.pendingMounts[:pos], u.pendingMounts[pos+1:]...)
+		return err
+	}
 	if isBlockDevice, err := u.drives.addInterface(s); err != nil {
 		return err
 	} else if isBlockDevice {
@@ -524,50 +525,3 @@
 
 	return blockDevice, nil
 }
-
-func (i InterfacesAndProperties) isMounted() bool {
-	propFS, ok := i[dbusFilesystemInterface]
-	if !ok {
-		return false
-	}
-	mountpointsVariant, ok := propFS["MountPoints"]
-	if !ok {
-		return false
-	}
-	if reflect.TypeOf(mountpointsVariant.Value).Kind() != reflect.Slice {
-		return false
-	}
-	mountpoints := reflect.ValueOf(mountpointsVariant.Value).Len()
-
-	return mountpoints > 0
-}
-
-func (i InterfacesAndProperties) hasPartition() bool {
-	prop, ok := i[dbusPartitionInterface]
-	if !ok {
-		return false
-	}
-	// check if a couple of properties exist
-	if _, ok := prop["UUID"]; !ok {
-		return false
-	}
-	if _, ok := prop["Table"]; !ok {
-		return false
-	}
-	return true
-}
-
-func (i InterfacesAndProperties) isPartitionable() bool {
-	prop, ok := i[dbusBlockInterface]
-	if !ok {
-		return false
-	}
-	partitionableHintVariant, ok := prop["HintPartitionable"]
-	if !ok {
-		return false
-	}
-	if reflect.TypeOf(partitionableHintVariant.Value).Kind() != reflect.Bool {
-		return false
-	}
-	return reflect.ValueOf(partitionableHintVariant.Value).Bool()
-}