src/device/media_transfer_protocol/media_transfer_protocol_manager.cc

   1 // Copyright (c) 2012 The Chromium Authors. All rights reserved.
   2 // Use of this source code is governed by a BSD-style license that can be
   3 // found in the LICENSE file.
   4
   5 #include "device/media_transfer_protocol/media_transfer_protocol_manager.h"
   6
   7 #include <algorithm>
   8 #include <map>
   9 #include <queue>
  10 #include <set>
  11 #include <utility>
  12
  13 #include "base/bind.h"
  14 #include "base/command_line.h"
  15 #include "base/location.h"
  16 #include "base/memory/weak_ptr.h"
  17 #include "base/observer_list.h"
  18 #include "base/sequenced_task_runner.h"
  19 #include "base/stl_util.h"
  20 #include "base/threading/thread_checker.h"
  21 #include "dbus/bus.h"
  22 #include "device/media_transfer_protocol/media_transfer_protocol_daemon_client.h"
  23 #include "device/media_transfer_protocol/mtp_file_entry.pb.h"
  24 #include "device/media_transfer_protocol/mtp_storage_info.pb.h"
  25 #include "third_party/cros_system_api/dbus/service_constants.h"
  26
  27 #if defined(OS_CHROMEOS)
  28 #include "chromeos/dbus/dbus_thread_manager.h"
  29 #endif
  30
  31 namespace device {
  32
  33 namespace {
  34
  35 MediaTransferProtocolManager* g_media_transfer_protocol_manager = NULL;
  36
  37 // When reading directory entries, this is the number of entries for
  38 // GetFileInfo() to read in one operation. If set too low, efficiency goes down
  39 // slightly due to the overhead of D-Bus calls. If set too high, then slow
  40 // devices may trigger a D-Bus timeout.
  41 // The value below is a good initial estimate.
  42 const size_t kFileInfoToFetchChunkSize = 25;
  43
  44 // On the first call to GetFileInfo, the offset to use is 0.
  45 const size_t kInitialOffset = 0;
  46
  47 // The MediaTransferProtocolManager implementation.
  48 class MediaTransferProtocolManagerImpl : public MediaTransferProtocolManager {
  49  public:
  50   explicit MediaTransferProtocolManagerImpl(
  51       scoped_refptr<base::SequencedTaskRunner> task_runner)
  52       : weak_ptr_factory_(this) {
  53 #if defined(OS_CHROMEOS)
  54     DCHECK(!task_runner.get());
  55 #else
  56     DCHECK(task_runner.get());
  57     dbus::Bus::Options options;
  58     options.bus_type = dbus::Bus::SYSTEM;
  59     options.connection_type = dbus::Bus::PRIVATE;
  60     options.dbus_task_runner = task_runner;
  61     session_bus_ = new dbus::Bus(options);
  62 #endif
  63
  64     if (GetBus()) {
  65       // Listen for future mtpd service owner changes, in case it is not
  66       // available right now. There is no guarantee on Linux or ChromeOS that
  67       // mtpd is running already.
  68       mtpd_owner_changed_callback_ = base::Bind(
  69           &MediaTransferProtocolManagerImpl::FinishSetupOnOriginThread,
  70           weak_ptr_factory_.GetWeakPtr());
  71       GetBus()->ListenForServiceOwnerChange(mtpd::kMtpdServiceName,
  72                                             mtpd_owner_changed_callback_);
  73       GetBus()->GetServiceOwner(mtpd::kMtpdServiceName,
  74                                 mtpd_owner_changed_callback_);
  75     }
  76   }
  77
  78   virtual ~MediaTransferProtocolManagerImpl() {
  79     DCHECK(g_media_transfer_protocol_manager);
  80     g_media_transfer_protocol_manager = NULL;
  81     if (GetBus()) {
  82       GetBus()->UnlistenForServiceOwnerChange(mtpd::kMtpdServiceName,
  83                                               mtpd_owner_changed_callback_);
  84     }
  85
  86 #if !defined(OS_CHROMEOS)
  87     session_bus_->GetDBusTaskRunner()->PostTask(
  88         FROM_HERE, base::Bind(&dbus::Bus::ShutdownAndBlock, session_bus_));
  89 #endif
  90
  91     VLOG(1) << "MediaTransferProtocolManager Shutdown completed";
  92   }
  93
  94   // MediaTransferProtocolManager override.
  95   virtual void AddObserver(Observer* observer) OVERRIDE {
  96     DCHECK(thread_checker_.CalledOnValidThread());
  97     observers_.AddObserver(observer);
  98   }
  99
 100   // MediaTransferProtocolManager override.
 101   virtual void RemoveObserver(Observer* observer) OVERRIDE {
 102     DCHECK(thread_checker_.CalledOnValidThread());
 103     observers_.RemoveObserver(observer);
 104   }
 105
 106   // MediaTransferProtocolManager override.
 107   virtual const std::vector<std::string> GetStorages() const OVERRIDE {
 108     DCHECK(thread_checker_.CalledOnValidThread());
 109     std::vector<std::string> storages;
 110     for (StorageInfoMap::const_iterator it = storage_info_map_.begin();
 111          it != storage_info_map_.end();
 112          ++it) {
 113       storages.push_back(it->first);
 114     }
 115     return storages;
 116   }
 117
 118   // MediaTransferProtocolManager override.
 119   virtual const MtpStorageInfo* GetStorageInfo(
 120       const std::string& storage_name) const OVERRIDE {
 121     DCHECK(thread_checker_.CalledOnValidThread());
 122     StorageInfoMap::const_iterator it = storage_info_map_.find(storage_name);
 123     return it != storage_info_map_.end() ? &it->second : NULL;
 124   }
 125
 126   // MediaTransferProtocolManager override.
 127   virtual void OpenStorage(const std::string& storage_name,
 128                            const std::string& mode,
 129                            const OpenStorageCallback& callback) OVERRIDE {
 130     DCHECK(thread_checker_.CalledOnValidThread());
 131     if (!ContainsKey(storage_info_map_, storage_name) || !mtp_client_) {
 132       callback.Run(std::string(), true);
 133       return;
 134     }
 135     open_storage_callbacks_.push(callback);
 136     mtp_client_->OpenStorage(
 137         storage_name,
 138         mode,
 139         base::Bind(&MediaTransferProtocolManagerImpl::OnOpenStorage,
 140                    weak_ptr_factory_.GetWeakPtr()),
 141         base::Bind(&MediaTransferProtocolManagerImpl::OnOpenStorageError,
 142                    weak_ptr_factory_.GetWeakPtr()));
 143   }
 144
 145   // MediaTransferProtocolManager override.
 146   virtual void CloseStorage(const std::string& storage_handle,
 147                             const CloseStorageCallback& callback) OVERRIDE {
 148     DCHECK(thread_checker_.CalledOnValidThread());
 149     if (!ContainsKey(handles_, storage_handle) || !mtp_client_) {
 150       callback.Run(true);
 151       return;
 152     }
 153     close_storage_callbacks_.push(std::make_pair(callback, storage_handle));
 154     mtp_client_->CloseStorage(
 155         storage_handle,
 156         base::Bind(&MediaTransferProtocolManagerImpl::OnCloseStorage,
 157                    weak_ptr_factory_.GetWeakPtr()),
 158         base::Bind(&MediaTransferProtocolManagerImpl::OnCloseStorageError,
 159                    weak_ptr_factory_.GetWeakPtr()));
 160   }
 161
 162   // MediaTransferProtocolManager override.
 163   virtual void ReadDirectory(const std::string& storage_handle,
 164                              uint32 file_id,
 165                              const ReadDirectoryCallback& callback) OVERRIDE {
 166     DCHECK(thread_checker_.CalledOnValidThread());
 167     if (!ContainsKey(handles_, storage_handle) || !mtp_client_) {
 168       callback.Run(std::vector<MtpFileEntry>(),
 169                    false /* no more entries */,
 170                    true /* error */);
 171       return;
 172     }
 173     read_directory_callbacks_.push(callback);
 174     mtp_client_->ReadDirectoryEntryIds(
 175         storage_handle,
 176         file_id,
 177         base::Bind(&MediaTransferProtocolManagerImpl::OnReadDirectoryEntryIds,
 178                    weak_ptr_factory_.GetWeakPtr(),
 179                    storage_handle),
 180         base::Bind(&MediaTransferProtocolManagerImpl::OnReadDirectoryError,
 181                    weak_ptr_factory_.GetWeakPtr()));
 182   }
 183
 184   // MediaTransferProtocolManager override.
 185   virtual void ReadFileChunk(const std::string& storage_handle,
 186                              uint32 file_id,
 187                              uint32 offset,
 188                              uint32 count,
 189                              const ReadFileCallback& callback) OVERRIDE {
 190     DCHECK(thread_checker_.CalledOnValidThread());
 191     if (!ContainsKey(handles_, storage_handle) || !mtp_client_) {
 192       callback.Run(std::string(), true);
 193       return;
 194     }
 195     read_file_callbacks_.push(callback);
 196     mtp_client_->ReadFileChunk(
 197         storage_handle, file_id, offset, count,
 198         base::Bind(&MediaTransferProtocolManagerImpl::OnReadFile,
 199                    weak_ptr_factory_.GetWeakPtr()),
 200         base::Bind(&MediaTransferProtocolManagerImpl::OnReadFileError,
 201                    weak_ptr_factory_.GetWeakPtr()));
 202   }
 203
 204   virtual void GetFileInfo(const std::string& storage_handle,
 205                            uint32 file_id,
 206                            const GetFileInfoCallback& callback) OVERRIDE {
 207     DCHECK(thread_checker_.CalledOnValidThread());
 208     if (!ContainsKey(handles_, storage_handle) || !mtp_client_) {
 209       callback.Run(MtpFileEntry(), true);
 210       return;
 211     }
 212     std::vector<uint32> file_ids;
 213     file_ids.push_back(file_id);
 214     get_file_info_callbacks_.push(callback);
 215     mtp_client_->GetFileInfo(
 216         storage_handle,
 217         file_ids,
 218         kInitialOffset,
 219         file_ids.size(),
 220         base::Bind(&MediaTransferProtocolManagerImpl::OnGetFileInfo,
 221                    weak_ptr_factory_.GetWeakPtr()),
 222         base::Bind(&MediaTransferProtocolManagerImpl::OnGetFileInfoError,
 223                    weak_ptr_factory_.GetWeakPtr()));
 224   }
 225
 226  private:
 227   // Map of storage names to storage info.
 228   typedef std::map<std::string, MtpStorageInfo> StorageInfoMap;
 229   // Callback queues - DBus communication is in-order, thus callbacks are
 230   // received in the same order as the requests.
 231   typedef std::queue<OpenStorageCallback> OpenStorageCallbackQueue;
 232   // (callback, handle)
 233   typedef std::queue<std::pair<CloseStorageCallback, std::string>
 234                     > CloseStorageCallbackQueue;
 235   typedef std::queue<ReadDirectoryCallback> ReadDirectoryCallbackQueue;
 236   typedef std::queue<ReadFileCallback> ReadFileCallbackQueue;
 237   typedef std::queue<GetFileInfoCallback> GetFileInfoCallbackQueue;
 238
 239   void OnStorageAttached(const std::string& storage_name) {
 240     DCHECK(thread_checker_.CalledOnValidThread());
 241     mtp_client_->GetStorageInfo(
 242         storage_name,
 243         base::Bind(&MediaTransferProtocolManagerImpl::OnGetStorageInfo,
 244                    weak_ptr_factory_.GetWeakPtr()),
 245         base::Bind(&base::DoNothing));
 246   }
 247
 248   void OnStorageDetached(const std::string& storage_name) {
 249     DCHECK(thread_checker_.CalledOnValidThread());
 250     if (storage_info_map_.erase(storage_name) == 0) {
 251       // This can happen for a storage where
 252       // MediaTransferProtocolDaemonClient::GetStorageInfo() failed.
 253       // Return to avoid giving observers phantom detach events.
 254       return;
 255     }
 256     FOR_EACH_OBSERVER(Observer,
 257                       observers_,
 258                       StorageChanged(false /* detach */, storage_name));
 259   }
 260
 261   void OnStorageChanged(bool is_attach, const std::string& storage_name) {
 262     DCHECK(thread_checker_.CalledOnValidThread());
 263     DCHECK(mtp_client_);
 264     if (is_attach)
 265       OnStorageAttached(storage_name);
 266     else
 267       OnStorageDetached(storage_name);
 268   }
 269
 270   void OnEnumerateStorages(const std::vector<std::string>& storage_names) {
 271     DCHECK(thread_checker_.CalledOnValidThread());
 272     DCHECK(mtp_client_);
 273     for (size_t i = 0; i < storage_names.size(); ++i) {
 274       if (ContainsKey(storage_info_map_, storage_names[i])) {
 275         // OnStorageChanged() might have gotten called first.
 276         continue;
 277       }
 278       OnStorageAttached(storage_names[i]);
 279     }
 280   }
 281
 282   void OnGetStorageInfo(const MtpStorageInfo& storage_info) {
 283     DCHECK(thread_checker_.CalledOnValidThread());
 284     const std::string& storage_name = storage_info.storage_name();
 285     if (ContainsKey(storage_info_map_, storage_name)) {
 286       // This should not happen, since MediaTransferProtocolManagerImpl should
 287       // only call EnumerateStorages() once, which populates |storage_info_map_|
 288       // with the already-attached devices.
 289       // After that, all incoming signals are either for new storage
 290       // attachments, which should not be in |storage_info_map_|, or for
 291       // storage detachments, which do not add to |storage_info_map_|.
 292       // Return to avoid giving observers phantom detach events.
 293       NOTREACHED();
 294       return;
 295     }
 296
 297     // New storage. Add it and let the observers know.
 298     storage_info_map_.insert(std::make_pair(storage_name, storage_info));
 299     FOR_EACH_OBSERVER(Observer,
 300                       observers_,
 301                       StorageChanged(true /* is attach */, storage_name));
 302   }
 303
 304   void OnOpenStorage(const std::string& handle) {
 305     DCHECK(thread_checker_.CalledOnValidThread());
 306     if (!ContainsKey(handles_, handle)) {
 307       handles_.insert(handle);
 308       open_storage_callbacks_.front().Run(handle, false);
 309     } else {
 310       NOTREACHED();
 311       open_storage_callbacks_.front().Run(std::string(), true);
 312     }
 313     open_storage_callbacks_.pop();
 314   }
 315
 316   void OnOpenStorageError() {
 317     open_storage_callbacks_.front().Run(std::string(), true);
 318     open_storage_callbacks_.pop();
 319   }
 320
 321   void OnCloseStorage() {
 322     DCHECK(thread_checker_.CalledOnValidThread());
 323     const std::string& handle = close_storage_callbacks_.front().second;
 324     if (ContainsKey(handles_, handle)) {
 325       handles_.erase(handle);
 326       close_storage_callbacks_.front().first.Run(false);
 327     } else {
 328       NOTREACHED();
 329       close_storage_callbacks_.front().first.Run(true);
 330     }
 331     close_storage_callbacks_.pop();
 332   }
 333
 334   void OnCloseStorageError() {
 335     DCHECK(thread_checker_.CalledOnValidThread());
 336     close_storage_callbacks_.front().first.Run(true);
 337     close_storage_callbacks_.pop();
 338   }
 339
 340   void OnReadDirectoryEntryIds(const std::string& storage_handle,
 341                                const std::vector<uint32>& file_ids) {
 342     DCHECK(thread_checker_.CalledOnValidThread());
 343
 344     if (file_ids.empty()) {
 345       OnGotDirectoryEntries(storage_handle,
 346                             file_ids,
 347                             kInitialOffset,
 348                             file_ids,
 349                             std::vector<MtpFileEntry>());
 350       return;
 351     }
 352
 353     std::vector<uint32> sorted_file_ids = file_ids;
 354     std::sort(sorted_file_ids.begin(), sorted_file_ids.end());
 355
 356     mtp_client_->GetFileInfo(
 357         storage_handle,
 358         file_ids,
 359         kInitialOffset,
 360         kFileInfoToFetchChunkSize,
 361         base::Bind(&MediaTransferProtocolManagerImpl::OnGotDirectoryEntries,
 362                    weak_ptr_factory_.GetWeakPtr(),
 363                    storage_handle,
 364                    file_ids,
 365                    kInitialOffset,
 366                    sorted_file_ids),
 367         base::Bind(&MediaTransferProtocolManagerImpl::OnReadDirectoryError,
 368                    weak_ptr_factory_.GetWeakPtr()));
 369   }
 370
 371   void OnGotDirectoryEntries(const std::string& storage_handle,
 372                              const std::vector<uint32>& file_ids,
 373                              size_t offset,
 374                              const std::vector<uint32>& sorted_file_ids,
 375                              const std::vector<MtpFileEntry>& file_entries) {
 376     DCHECK(thread_checker_.CalledOnValidThread());
 377     DCHECK_EQ(file_ids.size(), sorted_file_ids.size());
 378
 379     // Use |sorted_file_ids| to sanity check and make sure the results are a
 380     // subset of the requested file ids.
 381     for (size_t i = 0; i < file_entries.size(); ++i) {
 382       std::vector<uint32>::const_iterator it =
 383           std::lower_bound(sorted_file_ids.begin(),
 384                            sorted_file_ids.end(),
 385                            file_entries[i].item_id());
 386       if (it == sorted_file_ids.end()) {
 387         OnReadDirectoryError();
 388         return;
 389       }
 390     }
 391
 392     size_t next_offset = file_ids.size();
 393     if (offset < SIZE_MAX - kFileInfoToFetchChunkSize)
 394       next_offset = std::min(next_offset, offset + kFileInfoToFetchChunkSize);
 395     bool has_more = next_offset < file_ids.size();
 396     read_directory_callbacks_.front().Run(file_entries,
 397                                           has_more,
 398                                           false /* no error */);
 399     if (has_more) {
 400       mtp_client_->GetFileInfo(
 401           storage_handle,
 402           file_ids,
 403           next_offset,
 404           kFileInfoToFetchChunkSize,
 405           base::Bind(&MediaTransferProtocolManagerImpl::OnGotDirectoryEntries,
 406                      weak_ptr_factory_.GetWeakPtr(),
 407                      storage_handle,
 408                      file_ids,
 409                      next_offset,
 410                      sorted_file_ids),
 411           base::Bind(&MediaTransferProtocolManagerImpl::OnReadDirectoryError,
 412                      weak_ptr_factory_.GetWeakPtr()));
 413       return;
 414     }
 415     read_directory_callbacks_.pop();
 416   }
 417
 418   void OnReadDirectoryError() {
 419     DCHECK(thread_checker_.CalledOnValidThread());
 420     read_directory_callbacks_.front().Run(std::vector<MtpFileEntry>(),
 421                                           false /* no more entries */,
 422                                           true /* error */);
 423     read_directory_callbacks_.pop();
 424   }
 425
 426   void OnReadFile(const std::string& data) {
 427     DCHECK(thread_checker_.CalledOnValidThread());
 428     read_file_callbacks_.front().Run(data, false);
 429     read_file_callbacks_.pop();
 430   }
 431
 432   void OnReadFileError() {
 433     DCHECK(thread_checker_.CalledOnValidThread());
 434     read_file_callbacks_.front().Run(std::string(), true);
 435     read_file_callbacks_.pop();
 436   }
 437
 438   void OnGetFileInfo(const std::vector<MtpFileEntry>& entries) {
 439     DCHECK(thread_checker_.CalledOnValidThread());
 440     if (entries.size() == 1) {
 441       get_file_info_callbacks_.front().Run(entries[0], false /* no error */);
 442       get_file_info_callbacks_.pop();
 443     } else {
 444       OnGetFileInfoError();
 445     }
 446   }
 447
 448   void OnGetFileInfoError() {
 449     DCHECK(thread_checker_.CalledOnValidThread());
 450     get_file_info_callbacks_.front().Run(MtpFileEntry(), true);
 451     get_file_info_callbacks_.pop();
 452   }
 453
 454   // Get the Bus object used to communicate with mtpd.
 455   dbus::Bus* GetBus() {
 456     DCHECK(thread_checker_.CalledOnValidThread());
 457 #if defined(OS_CHROMEOS)
 458     return chromeos::DBusThreadManager::Get()->GetSystemBus();
 459 #else
 460     return session_bus_.get();
 461 #endif
 462   }
 463
 464   // Callback to finish initialization after figuring out if the mtpd service
 465   // has an owner, or if the service owner has changed.
 466   // |mtpd_service_owner| contains the name of the current owner, if any.
 467   void FinishSetupOnOriginThread(const std::string& mtpd_service_owner) {
 468     DCHECK(thread_checker_.CalledOnValidThread());
 469
 470     if (mtpd_service_owner == current_mtpd_owner_)
 471       return;
 472
 473     // In the case of a new service owner, clear |storage_info_map_|.
 474     // Assume all storages have been disconnected. If there is a new service
 475     // owner, reconnecting to it will reconnect all the storages as well.
 476
 477     // Save a copy of |storage_info_map_| keys as |storage_info_map_| can
 478     // change in OnStorageDetached().
 479     std::vector<std::string> storage_names;
 480     for (StorageInfoMap::const_iterator it = storage_info_map_.begin();
 481          it != storage_info_map_.end();
 482          ++it) {
 483       storage_names.push_back(it->first);
 484     }
 485     for (size_t i = 0; i != storage_names.size(); ++i)
 486       OnStorageDetached(storage_names[i]);
 487
 488     if (mtpd_service_owner.empty()) {
 489       current_mtpd_owner_.clear();
 490       mtp_client_.reset();
 491       return;
 492     }
 493
 494     current_mtpd_owner_ = mtpd_service_owner;
 495
 496     mtp_client_.reset(MediaTransferProtocolDaemonClient::Create(GetBus()));
 497
 498     // Set up signals and start initializing |storage_info_map_|.
 499     mtp_client_->ListenForChanges(
 500         base::Bind(&MediaTransferProtocolManagerImpl::OnStorageChanged,
 501                    weak_ptr_factory_.GetWeakPtr()));
 502     mtp_client_->EnumerateStorages(
 503         base::Bind(&MediaTransferProtocolManagerImpl::OnEnumerateStorages,
 504                    weak_ptr_factory_.GetWeakPtr()),
 505         base::Bind(&base::DoNothing));
 506   }
 507
 508   // Mtpd DBus client.
 509   scoped_ptr<MediaTransferProtocolDaemonClient> mtp_client_;
 510
 511 #if !defined(OS_CHROMEOS)
 512   // And a D-Bus session for talking to mtpd.
 513   scoped_refptr<dbus::Bus> session_bus_;
 514 #endif
 515
 516   // Device attachment / detachment observers.
 517   ObserverList<Observer> observers_;
 518
 519   // Map to keep track of attached storages by name.
 520   StorageInfoMap storage_info_map_;
 521
 522   // Set of open storage handles.
 523   std::set<std::string> handles_;
 524
 525   dbus::Bus::GetServiceOwnerCallback mtpd_owner_changed_callback_;
 526
 527   std::string current_mtpd_owner_;
 528
 529   // Queued callbacks.
 530   OpenStorageCallbackQueue open_storage_callbacks_;
 531   CloseStorageCallbackQueue close_storage_callbacks_;
 532   ReadDirectoryCallbackQueue read_directory_callbacks_;
 533   ReadFileCallbackQueue read_file_callbacks_;
 534   GetFileInfoCallbackQueue get_file_info_callbacks_;
 535
 536   base::ThreadChecker thread_checker_;
 537
 538   base::WeakPtrFactory<MediaTransferProtocolManagerImpl> weak_ptr_factory_;
 539
 540   DISALLOW_COPY_AND_ASSIGN(MediaTransferProtocolManagerImpl);
 541 };
 542
 543 }  // namespace
 544
 545 // static
 546 MediaTransferProtocolManager* MediaTransferProtocolManager::Initialize(
 547     scoped_refptr<base::SequencedTaskRunner> task_runner) {
 548   DCHECK(!g_media_transfer_protocol_manager);
 549
 550   g_media_transfer_protocol_manager =
 551       new MediaTransferProtocolManagerImpl(task_runner);
 552   VLOG(1) << "MediaTransferProtocolManager initialized";
 553
 554   return g_media_transfer_protocol_manager;
 555 }
 556
 557 }  // namespace device