00001
00002
00003
00004
00005
00006
00007
00008
00009
00010
00011
00012
00013
00014
00015
00016
00017
00018
00019
00020
00021
00022
00023
00024
00025
00026
00027
00028
00029
00030
00031
00032
00033
00034
00035
00036
00037
00038
00039 #include "BundleActions.h"
00040 #include "Bundle.h"
00041 #include "BundleDaemon.h"
00042 #include "BundleList.h"
00043 #include "conv_layers/ConvergenceLayer.h"
00044 #include "contacts/Link.h"
00045 #include "storage/BundleStore.h"
00046
00047 namespace dtn {
00048
00049
00050 void
00051 BundleActions::open_link(Link* link)
00052 {
00053 log_debug("opening link %s", link->name());
00054
00055 if (link->isopen()) {
00056 log_err("not opening link %s since already open", link->name());
00057 return;
00058 }
00059
00060 if (! link->isavailable()) {
00061 log_err("not opening link %s since not available", link->name());
00062 return;
00063 }
00064
00065 link->open();
00066 }
00067
00068
00069 void
00070 BundleActions::close_link(Link* link)
00071 {
00072 log_debug("closing link %s", link->name());
00073
00074 if (! link->isopen() && ! link->isopening()) {
00075 log_err("not closing link %s since not open", link->name());
00076 return;
00077 }
00078
00079 link->close();
00080 ASSERT(link->contact() == NULL);
00081 }
00082
00083
00084 bool
00085 BundleActions::send_bundle(Bundle* bundle, Link* link,
00086 ForwardingInfo::action_t action,
00087 const CustodyTimerSpec& custody_timer)
00088 {
00089 size_t total_len = BundleProtocol::formatted_length(bundle);
00090
00091 log_debug("send bundle *%p to %s link %s (%s) (total len %zu)",
00092 bundle, link->type_str(), link->name(), link->nexthop(),
00093 total_len);
00094
00095 if (link->state() != Link::OPEN) {
00096 log_err("send bundle *%p to %s link %s (%s): link not open!!",
00097 bundle, link->type_str(), link->name(), link->nexthop());
00098 return false;
00099 }
00100
00101 ForwardingInfo::state_t state = bundle->fwdlog_.get_latest_entry(link);
00102 if (state == ForwardingInfo::IN_FLIGHT) {
00103 log_err("send bundle *%p to %s link %s (%s): already in flight",
00104 bundle, link->type_str(), link->name(), link->nexthop());
00105 return false;
00106 }
00107
00108 if ((link->params().mtu_ != 0) && (total_len > link->params().mtu_)) {
00109 log_err("send bundle *%p to %s link %s (%s): length %zu > mtu %u",
00110 bundle, link->type_str(), link->name(), link->nexthop(),
00111 total_len, link->params().mtu_);
00112 return false;
00113 }
00114
00115 bundle->fwdlog_.add_entry(link, action, ForwardingInfo::IN_FLIGHT,
00116 custody_timer);
00117
00118 link->stats()->bundles_inflight_++;
00119 link->stats()->bytes_inflight_ += total_len;
00120
00121 ASSERT(link->contact() != NULL);
00122 link->clayer()->send_bundle(link->contact(), bundle);
00123 return true;
00124 }
00125
00126
00127 bool
00128 BundleActions::cancel_bundle(Bundle* bundle, Link* link)
00129 {
00130 log_debug("cancel bundle *%p on %s link %s (%s)",
00131 bundle, link->type_str(), link->name(), link->nexthop());
00132
00133 if (link->state() != Link::OPEN) {
00134 return false;
00135 }
00136
00137 ASSERT(link->contact() != NULL);
00138 bundle->fwdlog_.update(link, ForwardingInfo::CANCELLED);
00139
00140 return link->clayer()->cancel_bundle(link->contact(), bundle);
00141 }
00142
00143
00144 void
00145 BundleActions::inject_bundle(Bundle* bundle)
00146 {
00147 PANIC("XXX/demmer fix this");
00148
00149 log_debug("inject bundle *%p", bundle);
00150 BundleDaemon::instance()->pending_bundles()->push_back(bundle);
00151 store_add(bundle);
00152 }
00153
00154
00155 void
00156 BundleActions::store_add(Bundle* bundle)
00157 {
00158 log_debug("adding bundle %d to data store", bundle->bundleid_);
00159 bool added = BundleStore::instance()->add(bundle);
00160 if (! added) {
00161 log_crit("error adding bundle %d to data store!!", bundle->bundleid_);
00162 }
00163 }
00164
00165
00166 void
00167 BundleActions::store_update(Bundle* bundle)
00168 {
00169 log_debug("updating bundle %d in data store", bundle->bundleid_);
00170 bool updated = BundleStore::instance()->update(bundle);
00171 if (! updated) {
00172 log_crit("error updating bundle %d in data store!!", bundle->bundleid_);
00173 }
00174 }
00175
00176
00177 void
00178 BundleActions::store_del(Bundle* bundle)
00179 {
00180 log_debug("removing bundle %d from data store", bundle->bundleid_);
00181 bool removed = BundleStore::instance()->del(bundle->bundleid_);
00182 if (! removed) {
00183 log_crit("error adding bundle %d to data store!!", bundle->bundleid_);
00184 }
00185 }
00186
00187 }