import { SEND_MESSAGE, FINISH_SEND_MESSAGE, UPDATE_PRODUCER_FORM } from '../constants/action_types'
import { addHistory } from './producer_history'
const kafka = window.nodeRequire('no-kafka');

export function updateProducerForm(params){
  return {
    type: UPDATE_PRODUCER_FORM,
    params
  }
}

// @TODO params validation
export function sendMessage(params){
  //code mirror use ↵ to replace newline
  params.message = params.message.replace(/↵/g, '\n');
  let producer = new kafka.Producer({connectionString: params.url});
  return (dispatch) => {
    let paramsPayload = {
      url: params.url,
      message: params.message,
      topic: params.topic,
      partition: params.partition
    }
    dispatch(sendingMessageOnProgress(paramsPayload)); // message on progress
    dispatch(addHistory(paramsPayload));

    producer.init()
    .then(() => {
      return producer.send({
        partition: params.partition,
        topic: params.topic,
        message: { value: params.message}
      })
    })
    .then((result) =>{
      producer.end();
      let successMessage = `success with topic ${result[0].topic} offset ${result[0].offset}`
      return dispatch(finishSendMessage(
        { result: 'success', resultMessage: successMessage }
      ));
    })
    .catch((e) =>{
      producer.end();
      console.error(e);
      return dispatch(finishSendMessage({result: 'failed', resultMessage: e.message}));
    });
  };
}

export function sendingMessageOnProgress(params){
  return {
    type: SEND_MESSAGE,
    params
  }
}

export function finishSendMessage(params){ //success true or false
  return {
    type: FINISH_SEND_MESSAGE,
    status: 'sent',
    resultMessage: params.resultMessage,
    result: params.result
  }
}

@@ -813,17 +813,16 @@
 load));%0A
-%0A
     prod
@@ -1045,24 +1045,230 @@
 ucer.end();%0A
+      if(typeof result%5B0%5D.error == 'object')%7B%0A        console.error(result%5B0%5D.error);%0A        return dispatch(finishSendMessage(%7Bresult: 'failed', resultMessage: result%5B0%5D.error.message%7D));%0A      %7D else%7B%0A  
       let su
@@ -1349,16 +1349,18 @@
 ffset%7D%60%0A
+  
       re
@@ -1400,16 +1400,18 @@
         
+  
 %7B result
@@ -1461,20 +1461,30 @@
 %7D%0A      
+  
 ));%0A
+      %7D%0A
     %7D)%0A 
@@ -1497,24 +1497,25 @@
 tch((e) =%3E%7B%0A
+%0A
       produc

fix bug silent kafka error