media/mtransport/transportflow.cpp

   1 /* -*- Mode: C++; tab-width: 8; indent-tabs-mode: nil; c-basic-offset: 2 -*- */
   2 /* vim: set ts=2 et sw=2 tw=80: */
   3 /* This Source Code Form is subject to the terms of the Mozilla Public
   4  * License, v. 2.0. If a copy of the MPL was not distributed with this file,
   5  * You can obtain one at http://mozilla.org/MPL/2.0/. */
   6
   7 // Original author: ekr@rtfm.com
   8 #include <deque>
   9
  10 #include "logging.h"
  11 #include "runnable_utils.h"
  12 #include "transportflow.h"
  13 #include "transportlayer.h"
  14
  15 namespace mozilla {
  16
  17 MOZ_MTLOG_MODULE("mtransport")
  18
  19 NS_IMPL_ISUPPORTS0(TransportFlow)
  20
  21 // There are some hacks here to allow destruction off of
  22 // the main thread.
  23 TransportFlow::~TransportFlow() {
  24   // Make sure that if we are off the right thread, we have
  25   // no more attached signals.
  26   if (!CheckThreadInt()) {
  27     MOZ_ASSERT(SignalStateChange.is_empty());
  28     MOZ_ASSERT(SignalPacketReceived.is_empty());
  29   }
  30
  31   // Push the destruction onto the STS thread. Note that there
  32   // is still some possibility that someone is accessing this
  33   // object simultaneously, but as long as smart pointer discipline
  34   // is maintained, it shouldn't be possible to access and
  35   // destroy it simultaneously. The conversion to an nsAutoPtr
  36   // ensures automatic destruction of the queue at exit of
  37   // DestroyFinal.
  38   nsAutoPtr<std::deque<TransportLayer*> > layers_tmp(layers_.forget());
  39   RUN_ON_THREAD(target_,
  40                 WrapRunnableNM(&TransportFlow::DestroyFinal, layers_tmp),
  41                 NS_DISPATCH_NORMAL);
  42 }
  43
  44 void TransportFlow::DestroyFinal(nsAutoPtr<std::deque<TransportLayer *> > layers) {
  45   ClearLayers(layers);
  46 }
  47
  48 void TransportFlow::ClearLayers(std::queue<TransportLayer *>* layers) {
  49   while (!layers->empty()) {
  50     delete layers->front();
  51     layers->pop();
  52   }
  53 }
  54
  55 void TransportFlow::ClearLayers(std::deque<TransportLayer *>* layers) {
  56   while (!layers->empty()) {
  57     delete layers->front();
  58     layers->pop_front();
  59   }
  60 }
  61
  62 nsresult TransportFlow::PushLayer(TransportLayer *layer) {
  63   CheckThread();
  64   ScopedDeletePtr<TransportLayer> layer_tmp(layer);  // Destroy on failure.
  65
  66   // Don't allow pushes once we are in error state.
  67   if (state_ == TransportLayer::TS_ERROR) {
  68     MOZ_MTLOG(ML_ERROR, id_ + ": Can't call PushLayer in error state for flow");
  69     return NS_ERROR_FAILURE;
  70   }
  71
  72   nsresult rv = layer->Init();
  73   if (!NS_SUCCEEDED(rv)) {
  74     // Destroy the rest of the flow, because it's no longer in an acceptable
  75     // state.
  76     ClearLayers(layers_.get());
  77
  78     // Set ourselves to have failed.
  79     MOZ_MTLOG(ML_ERROR, id_ << ": Layer initialization failed; invalidating");
  80     StateChangeInt(TransportLayer::TS_ERROR);
  81
  82     return rv;
  83   }
  84   EnsureSameThread(layer);
  85
  86   TransportLayer *old_layer = layers_->empty() ? nullptr : layers_->front();
  87
  88   // Re-target my signals to the new layer
  89   if (old_layer) {
  90     old_layer->SignalStateChange.disconnect(this);
  91     old_layer->SignalPacketReceived.disconnect(this);
  92   }
  93   layers_->push_front(layer_tmp.forget());
  94   layer->Inserted(this, old_layer);
  95
  96   layer->SignalStateChange.connect(this, &TransportFlow::StateChange);
  97   layer->SignalPacketReceived.connect(this, &TransportFlow::PacketReceived);
  98   StateChangeInt(layer->state());
  99
 100   return NS_OK;
 101 }
 102
 103 // This is all-or-nothing.
 104 nsresult TransportFlow::PushLayers(nsAutoPtr<std::queue<TransportLayer *> > layers) {
 105   CheckThread();
 106
 107   MOZ_ASSERT(!layers->empty());
 108   if (layers->empty()) {
 109     MOZ_MTLOG(ML_ERROR, id_ << ": Can't call PushLayers with empty layers");
 110     return NS_ERROR_INVALID_ARG;
 111   }
 112
 113   // Don't allow pushes once we are in error state.
 114   if (state_ == TransportLayer::TS_ERROR) {
 115     MOZ_MTLOG(ML_ERROR,
 116               id_ << ": Can't call PushLayers in error state for flow ");
 117     ClearLayers(layers.get());
 118     return NS_ERROR_FAILURE;
 119   }
 120
 121   nsresult rv = NS_OK;
 122
 123   // Disconnect all the old signals.
 124   disconnect_all();
 125
 126   TransportLayer *layer;
 127
 128   while (!layers->empty()) {
 129     TransportLayer *old_layer = layers_->empty() ? nullptr : layers_->front();
 130     layer = layers->front();
 131
 132     rv = layer->Init();
 133     if (NS_FAILED(rv)) {
 134       MOZ_MTLOG(ML_ERROR,
 135                 id_ << ": Layer initialization failed; invalidating flow ");
 136       break;
 137     }
 138
 139     EnsureSameThread(layer);
 140
 141     // Push the layer onto the queue.
 142     layers_->push_front(layer);
 143     layers->pop();
 144     layer->Inserted(this, old_layer);
 145   }
 146
 147   if (NS_FAILED(rv)) {
 148     // Destroy any layers we could not push.
 149     ClearLayers(layers);
 150
 151     // Now destroy the rest of the flow, because it's no longer
 152     // in an acceptable state.
 153     ClearLayers(layers_);
 154
 155     // Set ourselves to have failed.
 156     StateChangeInt(TransportLayer::TS_ERROR);
 157
 158     // Return failure.
 159     return rv;
 160   }
 161
 162   // Finally, attach ourselves to the top layer.
 163   layer->SignalStateChange.connect(this, &TransportFlow::StateChange);
 164   layer->SignalPacketReceived.connect(this, &TransportFlow::PacketReceived);
 165   StateChangeInt(layer->state());  // Signals if the state changes.
 166
 167   return NS_OK;
 168 }
 169
 170 TransportLayer *TransportFlow::top() const {
 171   CheckThread();
 172
 173   return layers_->empty() ? nullptr : layers_->front();
 174 }
 175
 176 TransportLayer *TransportFlow::GetLayer(const std::string& id) const {
 177   CheckThread();
 178
 179   for (std::deque<TransportLayer *>::const_iterator it = layers_->begin();
 180        it != layers_->end(); ++it) {
 181     if ((*it)->id() == id)
 182       return *it;
 183   }
 184
 185   return nullptr;
 186 }
 187
 188 TransportLayer::State TransportFlow::state() {
 189   CheckThread();
 190
 191   return state_;
 192 }
 193
 194 TransportResult TransportFlow::SendPacket(const unsigned char *data,
 195                                           size_t len) {
 196   CheckThread();
 197
 198   if (state_ != TransportLayer::TS_OPEN) {
 199     return TE_ERROR;
 200   }
 201   return top() ? top()->SendPacket(data, len) : TE_ERROR;
 202 }
 203
 204 bool TransportFlow::Contains(TransportLayer *layer) const {
 205   if (layers_) {
 206     for (auto l = layers_->begin(); l != layers_->end(); ++l) {
 207       if (*l == layer) {
 208         return true;
 209       }
 210     }
 211   }
 212   return false;
 213 }
 214
 215 void TransportFlow::EnsureSameThread(TransportLayer *layer)  {
 216   // Enforce that if any of the layers have a thread binding,
 217   // they all have the same binding.
 218   if (target_) {
 219     const nsCOMPtr<nsIEventTarget>& lthread = layer->GetThread();
 220
 221     if (lthread && (lthread != target_))
 222       MOZ_CRASH();
 223   }
 224   else {
 225     target_ = layer->GetThread();
 226   }
 227 }
 228
 229 void TransportFlow::StateChangeInt(TransportLayer::State state) {
 230   CheckThread();
 231
 232   if (state == state_) {
 233     return;
 234   }
 235
 236   state_ = state;
 237   SignalStateChange(this, state_);
 238 }
 239
 240 void TransportFlow::StateChange(TransportLayer *layer,
 241                                 TransportLayer::State state) {
 242   CheckThread();
 243
 244   StateChangeInt(state);
 245 }
 246
 247 void TransportFlow::PacketReceived(TransportLayer* layer,
 248                                    const unsigned char *data,
 249                                    size_t len) {
 250   CheckThread();
 251
 252   SignalPacketReceived(this, data, len);
 253 }
 254
 255 }  // close namespace