PROTON-1311: C interface to get/set the link max-msessage-size
diff --git a/proton-c/bindings/python/proton/__init__.py b/proton-c/bindings/python/proton/__init__.py
index 516daf1..d3f6922 100644
--- a/proton-c/bindings/python/proton/__init__.py
+++ b/proton-c/bindings/python/proton/__init__.py
@@ -2868,6 +2868,16 @@
   def drained(self):
     return pn_link_drained(self._impl)
 
+  @property
+  def remote_max_message_size(self):
+    return pn_link_remote_max_message_size(self._impl)
+
+  def _get_max_message_size(self):
+    return pn_link_max_message_size(self._impl)
+  def _set_max_message_size(self, mode):
+    pn_link_set_max_message_size(self._impl, mode)
+  max_message_size = property(_get_max_message_size, _set_max_message_size)
+
   def detach(self):
     return pn_link_detach(self._impl)
 
diff --git a/proton-c/include/proton/link.h b/proton-c/include/proton/link.h
index 8ec162e..d52e6e7 100644
--- a/proton-c/include/proton/link.h
+++ b/proton-c/include/proton/link.h
@@ -647,6 +647,36 @@
 PN_EXTERN bool pn_link_draining(pn_link_t *receiver);
 
 /**
+ * **Experimental** - Get the maximum message size for a link.
+ *
+ * A zero maximum message size means the size is unlimited.
+ *
+ * @param[in] link a link object
+ * @return the maximum message size for a link.
+ */
+PN_EXTERN uint64_t pn_link_max_message_size(pn_link_t *link);
+
+/**
+ * **Experimental** - Set the maximum message size for a link.
+ *
+ * A zero maximum message size means the size is unlimited.
+ *
+ * @param[in] link a link object
+ * @param[in] size the maximum message size for the link
+ */
+PN_EXTERN void pn_link_set_max_message_size(pn_link_t *link, uint64_t size);
+
+/**
+ * **Experimental** - Get the remote view of the maximum message size for a link.
+ *
+ * A zero maximum message size means the size is unlimited.
+ *
+ * @param[in] link a link object
+ * @return the remote view of the maximum message size for a link
+ */
+PN_EXTERN uint64_t pn_link_remote_max_message_size(pn_link_t *link);
+
+/**
  * @}
  */
 
diff --git a/proton-c/src/core/engine-internal.h b/proton-c/src/core/engine-internal.h
index fdaf272..1dbe91c 100644
--- a/proton-c/src/core/engine-internal.h
+++ b/proton-c/src/core/engine-internal.h
@@ -291,6 +291,8 @@
   pn_delivery_t *current;
   pn_record_t *context;
   size_t unsettled_count;
+  uint64_t max_message_size;
+  uint64_t remote_max_message_size;
   pn_sequence_t available;
   pn_sequence_t credit;
   pn_sequence_t queued;
diff --git a/proton-c/src/core/engine.c b/proton-c/src/core/engine.c
index 99d311b..8c2aeb0 100644
--- a/proton-c/src/core/engine.c
+++ b/proton-c/src/core/engine.c
@@ -1176,6 +1176,8 @@
   pni_terminus_init(&link->remote_target, PN_UNSPECIFIED);
   link->unsettled_head = link->unsettled_tail = link->current = NULL;
   link->unsettled_count = 0;
+  link->max_message_size = 0;
+  link->remote_max_message_size = 0;
   link->available = 0;
   link->credit = 0;
   link->queued = 0;
@@ -1956,6 +1958,21 @@
   return receiver->drain && (pn_link_credit(receiver) > pn_link_queued(receiver));
 }
 
+uint64_t pn_link_max_message_size(pn_link_t *link)
+{
+  return link->max_message_size;
+}
+
+void pn_link_set_max_message_size(pn_link_t *link, uint64_t size)
+{
+  link->max_message_size = size;
+}
+
+uint64_t pn_link_remote_max_message_size(pn_link_t *link)
+{
+  return link->remote_max_message_size;
+}
+
 pn_link_t *pn_delivery_link(pn_delivery_t *delivery)
 {
   assert(delivery);
diff --git a/proton-c/src/core/transport.c b/proton-c/src/core/transport.c
index e3b88b1..444145a 100644
--- a/proton-c/src/core/transport.c
+++ b/proton-c/src/core/transport.c
@@ -1342,13 +1342,14 @@
   pn_bytes_t dist_mode;
   bool snd_settle, rcv_settle;
   uint8_t snd_settle_mode, rcv_settle_mode;
-  int err = pn_data_scan(args, "D.[SIo?B?BD.[SIsIo.s]D.[SIsIo]..I]", &name, &handle,
+  uint64_t max_msgsz;
+  int err = pn_data_scan(args, "D.[SIo?B?BD.[SIsIo.s]D.[SIsIo]..IL]", &name, &handle,
                          &is_sender,
                          &snd_settle, &snd_settle_mode,
                          &rcv_settle, &rcv_settle_mode,
                          &source, &src_dr, &src_exp, &src_timeout, &src_dynamic, &dist_mode,
                          &target, &tgt_dr, &tgt_exp, &tgt_timeout, &tgt_dynamic,
-                         &idc);
+                         &idc, &max_msgsz);
   if (err) return err;
   char strbuf[128];      // avoid malloc for most link names
   char *strheap = (name.size >= sizeof(strbuf)) ? (char *) malloc(name.size + 1) : NULL;
@@ -1444,6 +1445,10 @@
     link->state.delivery_count = idc;
   }
 
+  if (max_msgsz) {
+    link->remote_max_message_size = max_msgsz;
+  }
+
   pn_collector_put(transport->connection->collector, PN_OBJECT, link, PN_LINK_REMOTE_OPEN);
   return 0;
 }
@@ -1988,7 +1993,7 @@
         if (err) return err;
       } else {
         int err = pn_post_frame(transport, AMQP_FRAME_TYPE, ssn_state->local_channel,
-                                "DL[SIoBB?DL[SIsIoC?sCnCC]?DL[SIsIoCC]nnI]", ATTACH,
+                                "DL[SIoBB?DL[SIsIoC?sCnCC]?DL[SIsIoCC]nnIL]", ATTACH,
                                 pn_string_get(link->name),
                                 state->local_handle,
                                 endpoint->type == RECEIVER,
@@ -2013,7 +2018,7 @@
                                 link->target.dynamic,
                                 link->target.properties,
                                 link->target.capabilities,
-                                0);
+                                0, link->max_message_size);
         if (err) return err;
       }
     }
diff --git a/tests/python/proton_tests/engine.py b/tests/python/proton_tests/engine.py
index d011210..497bb7d 100644
--- a/tests/python/proton_tests/engine.py
+++ b/tests/python/proton_tests/engine.py
@@ -746,6 +746,17 @@
     assert self.snd.remote_rcv_settle_mode == Link.RCV_SECOND
     assert self.rcv.remote_snd_settle_mode == Link.SND_UNSETTLED
 
+  def test_max_message_size(self):
+    if "java" in sys.platform:
+      raise Skipped()
+    assert self.snd.max_message_size == 0
+    assert self.rcv.remote_max_message_size == 0
+    self.snd.max_message_size = 13579
+    self.snd.open()
+    self.rcv.open()
+    self.pump()
+    assert self.rcv.remote_max_message_size == 13579
+
   def test_cleanup(self):
     snd, rcv = self.link("test-link")
     snd.open()